<?php

declare(strict_types=1);

namespace Drupal\flowdrop_session\Service;

use Drupal\Component\Serialization\Json;
use Drupal\Core\Database\Connection;
use Drupal\Core\Entity\EntityStorageInterface;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Logger\LoggerChannelFactoryInterface;
use Drupal\Core\Session\AccountInterface;
use Drupal\flowdrop_session\Constants\MessageAnnotation;
use Drupal\flowdrop_session\Constants\MessageRole;
use Drupal\flowdrop_session\Constants\MessageStatus;
use Drupal\flowdrop_session\Constants\SessionStatus;
use Drupal\flowdrop_session\Entity\FlowDropSessionMessage;
use Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface;
use Drupal\flowdrop_pipeline\Service\PipelineLineageResolver;
use Drupal\flowdrop_session\Entity\FlowDropSession;
use Drupal\flowdrop_session\Entity\FlowDropSessionInterface;
use Drupal\flowdrop_workflow\WorkflowDefinitionInterface;
use Psr\Log\LoggerInterface;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;

/**
 * Service for managing sessions and message execution.
 *
 * This service handles the core business logic for:
 * - Creating and managing sessions
 * - Processing user messages and triggering workflow execution
 * - Managing the chat-like interface between user and workflow.
 *
 * @internal This class is not part of the public API and may change without notice.
 */
class SessionService implements SessionServiceInterface {

  /**
   * The entity type manager.
   *
   * @var \Drupal\Core\Entity\EntityTypeManagerInterface
   */
  protected EntityTypeManagerInterface $entityTypeManager;

  /**
   * The current user.
   *
   * @var \Drupal\Core\Session\AccountInterface
   */
  protected AccountInterface $currentUser;

  /**
   * The logger.
   *
   * @var \Psr\Log\LoggerInterface
   */
  protected LoggerInterface $logger;

  /**
   * The event dispatcher.
   *
   * @var \Symfony\Component\EventDispatcher\EventDispatcherInterface
   */
  protected EventDispatcherInterface $eventDispatcher;

  /**
   * The session entity storage.
   *
   * @var \Drupal\Core\Entity\EntityStorageInterface
   */
  protected EntityStorageInterface $sessionStorage;

  /**
   * The message entity storage.
   *
   * @var \Drupal\Core\Entity\EntityStorageInterface
   */
  protected EntityStorageInterface $messageStorage;

  /**
   * The database connection.
   *
   * @var \Drupal\Core\Database\Connection
   */
  protected Connection $database;

  /**
   * Resolves pipeline lineage (root / parent) for message execution ids.
   */
  protected PipelineLineageResolver $lineageResolver;

  /**
   * Constructs a new SessionService.
   *
   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entity_type_manager
   *   The entity type manager.
   * @param \Drupal\Core\Session\AccountInterface $current_user
   *   The current user.
   * @param \Drupal\Core\Logger\LoggerChannelFactoryInterface $logger_factory
   *   The logger factory.
   * @param \Symfony\Component\EventDispatcher\EventDispatcherInterface $event_dispatcher
   *   The event dispatcher.
   * @param \Drupal\Core\Database\Connection $database
   *   The database connection.
   * @param \Drupal\flowdrop_pipeline\Service\PipelineLineageResolver $lineage_resolver
   *   Resolves pipeline lineage for message execution ids.
   */
  public function __construct(
    EntityTypeManagerInterface $entity_type_manager,
    AccountInterface $current_user,
    LoggerChannelFactoryInterface $logger_factory,
    EventDispatcherInterface $event_dispatcher,
    Connection $database,
    PipelineLineageResolver $lineage_resolver,
  ) {
    $this->entityTypeManager = $entity_type_manager;
    $this->currentUser = $current_user;
    $this->logger = $logger_factory->get('flowdrop_session');
    $this->eventDispatcher = $event_dispatcher;
    $this->database = $database;
    $this->lineageResolver = $lineage_resolver;
    $this->sessionStorage = $entity_type_manager->getStorage('flowdrop_session');
    $this->messageStorage = $entity_type_manager->getStorage('flowdrop_session_message');
  }

  /**
   * Creates a new session for a workflow.
   *
   * @param \Drupal\flowdrop_workflow\WorkflowDefinitionInterface $workflow
   *   The workflow entity.
   * @param string|null $name
   *   Optional session name. If not provided or if it matches the "Session N"
   *   pattern (which may be stale from frontend), a unique name is generated.
   * @param array<string, mixed> $metadata
   *   Optional metadata for the session.
   *
   * @return \Drupal\flowdrop_session\Entity\FlowDropSessionInterface
   *   The created session entity.
   *
   * @throws \Drupal\Core\Entity\EntityStorageException
   *   If the entity cannot be saved.
   */
  public function createSession(
    WorkflowDefinitionInterface $workflow,
    ?string $name = NULL,
    array $metadata = [],
  ): FlowDropSessionInterface {
    // Auto-generate a name when none is given or the frontend sent a stale
    // "Session N" placeholder; keep an explicit, custom name as-is. The
    // session list appends "#<id>" to every row for disambiguation, so the
    // stored name only needs the workflow label — no post-save round-trip to
    // fold the ID into the name.
    $shouldGenerate = $name === NULL || preg_match('/^Session \d+$/', $name) === 1;

    $session = FlowDropSession::create([
      'workflow_id' => $workflow->id(),
      'name' => $shouldGenerate ? (string) $workflow->label() : $name,
      'status' => SessionStatus::IDLE,
      'uid' => $this->currentUser->id(),
      // JSON-encode metadata for string_long field type.
      // Passing an array directly would be cast to the string "Array".
      'metadata' => Json::encode($metadata),
    ]);

    $session->save();

    $this->logger->info('Created session @id for workflow @workflow', [
      '@id' => $session->id(),
      '@workflow' => $workflow->id(),
    ]);

    return $session;
  }

  /**
   * Gets sessions for a workflow.
   *
   * @param string $workflow_id
   *   The workflow ID.
   * @param int|null $user_id
   *   Optional user ID to filter by owner.
   * @param int $limit
   *   Maximum number of sessions to return.
   * @param int $offset
   *   Number of sessions to skip.
   *
   * @return array<\Drupal\flowdrop_session\Entity\FlowDropSessionInterface>
   *   Array of session entities.
   */
  public function getSessionsForWorkflow(
    string $workflow_id,
    ?int $user_id = NULL,
    int $limit = 20,
    int $offset = 0,
  ): array {
    $query = $this->sessionStorage->getQuery()
      ->condition('workflow_id', $workflow_id)
      ->accessCheck(TRUE)
      ->sort('created', 'DESC')
      ->range($offset, $limit);

    if ($user_id !== NULL) {
      $query->condition('uid', $user_id);
    }

    $ids = $query->execute();

    if (empty($ids)) {
      return [];
    }

    $sessions = $this->sessionStorage->loadMultiple($ids);
    $result = [];
    foreach ($sessions as $session) {
      if ($session instanceof FlowDropSessionInterface) {
        $result[] = $session;
      }
    }

    return $result;
  }

  /**
   * Gets a session by ID.
   *
   * @param int $session_id
   *   The session ID.
   *
   * @return \Drupal\flowdrop_session\Entity\FlowDropSessionInterface|null
   *   The session entity or NULL if not found.
   */
  public function getSession(int $session_id): ?FlowDropSessionInterface {
    $entity = $this->sessionStorage->load($session_id);
    return $entity instanceof FlowDropSessionInterface ? $entity : NULL;
  }

  /**
   * Checks if a workflow has a chat input node.
   *
   * Chat input nodes allow users to type messages in the playground.
   * Workflows without chat input nodes should hide the text input and
   * only show the Run button.
   *
   * @param \Drupal\flowdrop_workflow\WorkflowDefinitionInterface $workflow
   *   The workflow entity.
   *
   * @return bool
   *   TRUE if the workflow has a chat input node, FALSE otherwise.
   */
  public function hasChatInputNode(WorkflowDefinitionInterface $workflow): bool {
    return $this->findChatInputNode($workflow) !== NULL;
  }

  /**
   * Deletes a session and all its messages.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session to delete.
   *
   * @throws \Drupal\Core\Entity\EntityStorageException
   *   If deletion fails.
   */
  public function deleteSession(FlowDropSessionInterface $session): void {
    $session_id = $session->id();
    if ($session_id === NULL) {
      return;
    }

    // Delete all messages for this session.
    $message_ids = $this->messageStorage->getQuery()
      ->condition('session_id', $session_id)
      ->accessCheck(TRUE)
      ->execute();

    if (!empty($message_ids)) {
      $messages = $this->messageStorage->loadMultiple($message_ids);
      $this->messageStorage->delete($messages);
    }

    // Delete the session.
    $session->delete();

    $this->logger->info('Deleted session @id with @count messages', [
      '@id' => $session_id,
      '@count' => count($message_ids),
    ]);
  }

  /**
   * Gets messages for a session.
   *
   * Supports three cursor modes for large sessions. Exactly one of
   * $after_sequence / $before_sequence / $latest should be meaningful at a
   * time; if more than one is given, $before_sequence wins over $latest,
   * and $after_sequence is applied independently as a lower bound.
   *
   * - Forward (default, or $after_sequence): the oldest page, or the page
   *   immediately newer than $after_sequence. Used by live polling to tail
   *   new messages.
   * - Latest ($latest = TRUE): the newest $limit messages — the right
   *   initial load for a long session so the user lands at the bottom.
   * - Backward ($before_sequence): the $limit messages with the highest
   *   sequence numbers strictly below $before_sequence — used to lazy-load
   *   older history as the user scrolls up.
   *
   * The returned array is always ascending by sequence_number regardless of
   * mode, so callers render it top-to-bottom without re-sorting.
   *
   * @param int $session_id
   *   The session ID.
   * @param int|null $after_sequence
   *   Forward cursor — only messages with sequence_number strictly greater
   *   than this value.
   * @param int $limit
   *   Maximum number of messages to return.
   * @param int|null $before_sequence
   *   Backward cursor — only messages with sequence_number strictly less
   *   than this value, taking the highest such numbers.
   * @param bool $latest
   *   When TRUE (and no $before_sequence), return the newest $limit
   *   messages.
   *
   * @return array<\Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface>
   *   Array of message entities, ascending by sequence_number.
   */
  public function getMessages(
    int $session_id,
    ?int $after_sequence = NULL,
    int $limit = 100,
    ?int $before_sequence = NULL,
    bool $latest = FALSE,
  ): array {
    // Backward / latest modes anchor on the END of the range, so query
    // DESC to grab the highest sequence numbers, then flip to ascending
    // for the response. Forward / default stay ASC from the low end.
    $descend = $before_sequence !== NULL || $latest;

    $query = $this->messageStorage->getQuery()
      ->condition('session_id', $session_id)
      ->accessCheck(TRUE)
      ->sort('sequence_number', $descend ? 'DESC' : 'ASC')
      ->range(0, $limit);

    if ($after_sequence !== NULL) {
      $query->condition('sequence_number', $after_sequence, '>');
    }
    if ($before_sequence !== NULL) {
      $query->condition('sequence_number', $before_sequence, '<');
    }

    $ids = $query->execute();
    if (empty($ids)) {
      return [];
    }

    $messages = $this->messageStorage->loadMultiple($ids);
    $result = [];
    foreach ($messages as $message) {
      if ($message instanceof FlowDropSessionMessageInterface) {
        $result[] = $message;
      }
    }

    // loadMultiple() keys by entity id and does not honor the query sort,
    // so order explicitly. Ascending by sequence_number is the contract
    // for every mode.
    usort(
      $result,
      static fn (FlowDropSessionMessageInterface $a, FlowDropSessionMessageInterface $b): int =>
        $a->getSequenceNumber() <=> $b->getSequenceNumber(),
    );

    return $result;
  }

  /**
   * Checks whether any message exists below a given sequence number.
   *
   * Powers the `hasOlder` response flag for backward pagination: given the
   * lowest sequence number on the current page, tells the client whether
   * scrolling up will yield more history. A single COUNT-bounded lookup,
   * cheaper than fetching a speculative extra page.
   *
   * @param int $session_id
   *   The session ID.
   * @param int $sequence
   *   The exclusive upper bound — looks for messages with sequence_number
   *   strictly less than this.
   *
   * @return bool
   *   TRUE when at least one older message exists.
   */
  public function hasMessagesBefore(int $session_id, int $sequence): bool {
    $ids = $this->messageStorage->getQuery()
      ->condition('session_id', $session_id)
      ->condition('sequence_number', $sequence, '<')
      ->accessCheck(TRUE)
      ->range(0, 1)
      ->execute();
    return !empty($ids);
  }

  /**
   * Adds a user message to the session.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session entity.
   * @param string $content
   *   The message content.
   * @param array<string, mixed> $metadata
   *   Optional metadata.
   *
   * @return \Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface
   *   The created message entity.
   *
   * @throws \Drupal\Core\Entity\EntityStorageException
   *   If the entity cannot be saved.
   */
  public function addUserMessage(
    FlowDropSessionInterface $session,
    string $content,
    array $metadata = [],
  ): FlowDropSessionMessageInterface {
    return $this->createMessage($session, MessageRole::USER, $content, NULL, $metadata);
  }

  /**
   * Adds an assistant message to the session.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session entity.
   * @param string $content
   *   The message content.
   * @param string|null $node_id
   *   Optional node ID that generated this message.
   * @param array<string, mixed> $metadata
   *   Optional metadata.
   *
   * @return \Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface
   *   The created message entity.
   *
   * @throws \Drupal\Core\Entity\EntityStorageException
   *   If the entity cannot be saved.
   */
  public function addAssistantMessage(
    FlowDropSessionInterface $session,
    string $content,
    ?string $node_id = NULL,
    array $metadata = [],
  ): FlowDropSessionMessageInterface {
    return $this->createMessage($session, MessageRole::ASSISTANT, $content, $node_id, $metadata);
  }

  /**
   * Adds a system message to the session.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session entity.
   * @param string $content
   *   The message content.
   * @param array<string, mixed> $metadata
   *   Optional metadata.
   * @param string|null $executionId
   *   Optional execution ID to link the message to a pipeline run.
   *
   * @return \Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface
   *   The created message entity.
   *
   * @throws \Drupal\Core\Entity\EntityStorageException
   *   If the entity cannot be saved.
   */
  public function addSystemMessage(
    FlowDropSessionInterface $session,
    string $content,
    array $metadata = [],
    ?string $executionId = NULL,
  ): FlowDropSessionMessageInterface {
    $message = $this->createMessage($session, MessageRole::SYSTEM, $content, NULL, $metadata);
    if ($executionId !== NULL) {
      $message->setExecutionId($executionId)->save();
    }
    return $message;
  }

  /**
   * Adds a log message to the session.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session entity.
   * @param string $content
   *   The log message content.
   * @param string|null $node_id
   *   The node ID this log relates to.
   * @param string $level
   *   The log level (info, warning, error, debug).
   * @param array<string, mixed> $metadata
   *   Optional additional metadata.
   * @param string|null $executionId
   *   Optional execution ID to link the message to a pipeline run.
   *
   * @return \Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface
   *   The created message entity.
   *
   * @throws \Drupal\Core\Entity\EntityStorageException
   *   If the entity cannot be saved.
   */
  public function addLogMessage(
    FlowDropSessionInterface $session,
    string $content,
    ?string $node_id = NULL,
    string $level = 'info',
    array $metadata = [],
    ?string $executionId = NULL,
  ): FlowDropSessionMessageInterface {
    $metadata['level'] = $level;
    $message = $this->createMessage($session, MessageRole::LOG, $content, $node_id, $metadata);
    if ($executionId !== NULL) {
      $message->setExecutionId($executionId)->save();
    }
    return $message;
  }

  /**
   * Creates a message entity.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session entity.
   * @param string $role
   *   The message role.
   * @param string $content
   *   The message content.
   * @param string|null $node_id
   *   Optional node ID.
   * @param array<string, mixed> $metadata
   *   Optional metadata.
   *
   * @return \Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface
   *   The created message entity.
   *
   * @throws \Drupal\Core\Entity\EntityStorageException
   *   If the entity cannot be saved.
   */
  protected function createMessage(
    FlowDropSessionInterface $session,
    string $role,
    string $content,
    ?string $node_id = NULL,
    array $metadata = [],
  ): FlowDropSessionMessageInterface {
    // Extract sequence_number from metadata if provided.
    $sequenceNumber = $metadata['sequence_number'] ?? NULL;
    unset($metadata['sequence_number']);

    // For user messages, store the current user's display name in metadata.
    // This allows the frontend to show the actual user name instead of "You".
    if ($role === MessageRole::USER && !isset($metadata['userName'])) {
      $metadata['userName'] = $this->currentUser->getDisplayName();
    }

    // Encode metadata as JSON string - the field is string_long type.
    $metadataJson = !empty($metadata) ? Json::encode($metadata) : '{}';

    // Determine initial status based on message role and type.
    // User messages start as 'pending' since they trigger workflow execution.
    // Interrupt request messages must also be 'pending' so the frontend knows
    // the user still needs to interact with them (buttons should be enabled).
    // Other system/assistant/log messages are 'completed' immediately since
    // they don't require processing and shouldn't block the UI.
    $isInterruptRequest = ($metadata['type'] ?? NULL) === 'interrupt_request';
    if ($role === MessageRole::USER || $isInterruptRequest) {
      $initialStatus = MessageStatus::PENDING;
    }
    else {
      $initialStatus = MessageStatus::COMPLETED;
    }

    // The entity's auto-increment id is the monotonic per-session ordering
    // signal. Reusing it as sequence_number means concurrent writers (a
    // user typing while a sub-pipeline fires on the same playground
    // session, async cron, etc.) can never collide on a value — InnoDB's
    // AUTO_INCREMENT is already serialized at the storage layer. Saves
    // one round-trip vs. SELECT MAX FOR UPDATE + INSERT, and removes the
    // lock contention that approach would create on a busy session.
    //
    // The transaction wraps INSERT and the sequence-stamp UPDATE so a
    // concurrent reader never observes the brief gap where sequence_number
    // still has the field default of 0. Nested startTransaction() folds
    // into a savepoint when sendMessage already opened one.
    $transaction = $this->database->startTransaction();
    try {
      $message = FlowDropSessionMessage::create([
        'session_id' => $session->id(),
        'role' => $role,
        'content' => $content,
        'node_id' => $node_id,
        'metadata' => $metadataJson,
        'status' => $initialStatus,
      ]);
      $message->save();

      // Adopt the freshly-assigned entity id as the sequence number unless
      // the caller pinned one explicitly via metadata['sequence_number'].
      if ($sequenceNumber === NULL) {
        $sequenceNumber = (int) $message->id();
      }
      $this->database->update($message->getEntityType()->getBaseTable())
        ->fields(['sequence_number' => (int) $sequenceNumber])
        ->condition('id', $message->id())
        ->execute();
      $message->setSequenceNumber((int) $sequenceNumber);

      // Update session's changed timestamp.
      $session->save();

      return $message;
    }
    catch (\Throwable $e) {
      $transaction->rollBack();
      throw $e;
    }
  }

  /**
   * Appends a message to the session conversation.
   *
   * Posting is a plain "append to the room" operation: it persists the
   * message and returns. It does not validate sequence ordering, flip the
   * session into RUNNING, or launch a pipeline. Use the workflow launcher
   * (`/api/flowdrop/workflow/{id}/run`) to start execution.
   *
   * The message is created with status COMPLETED — a posted message
   * requires no processing of its own.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session entity.
   * @param string $content
   *   The message content.
   * @param string $role
   *   The message role (defaults to USER).
   * @param array<string, mixed> $metadata
   *   Optional metadata to attach to the message.
   *
   * @return \Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface
   *   The created message entity.
   *
   * @throws \Drupal\Core\Entity\EntityStorageException
   *   If the entity cannot be saved.
   */
  public function postMessage(
    FlowDropSessionInterface $session,
    string $content,
    string $role = MessageRole::USER,
    array $metadata = [],
  ): FlowDropSessionMessageInterface {
    $message = $this->createMessage($session, $role, $content, NULL, $metadata);
    if (!in_array($message->getStatus(), MessageStatus::terminalStatuses(), TRUE)) {
      $message->setStatus(MessageStatus::COMPLETED)->save();
    }
    return $message;
  }

  /**
   * Stops the execution of a session.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session entity.
   *
   * @throws \RuntimeException
   *   If no execution is running.
   */
  public function stopExecution(FlowDropSessionInterface $session): void {
    if (!$session->isRunning()) {
      throw new \RuntimeException('No execution is running for this session.');
    }

    $pipeline_id = $session->getCurrentPipelineId();

    if ($pipeline_id !== NULL) {
      try {
        $pipeline_storage = $this->entityTypeManager->getStorage('flowdrop_pipeline');
        $pipeline = $pipeline_storage->load($pipeline_id);

        if ($pipeline !== NULL && method_exists($pipeline, 'markAsCancelled')) {
          $pipeline->markAsCancelled();
          $pipeline->save();
        }
      }
      catch (\Exception $e) {
        $this->logger->warning('Failed to cancel pipeline @id: @message', [
          '@id' => $pipeline_id,
          '@message' => $e->getMessage(),
        ]);
      }
    }

    // Update session status.
    $session->setStatus(SessionStatus::IDLE);
    $session->setCurrentPipelineId(NULL);
    $session->save();

    // Add system message.
    $this->addSystemMessage($session, 'Execution stopped by user.');

    $this->logger->info('Stopped execution for session @id', [
      '@id' => $session->id(),
    ]);
  }

  /**
   * Resets a session to idle state, recovering from stuck states.
   *
   * This method is useful when a session gets stuck due to failed executions
   * that didn't properly cleanup, network issues during processing, or
   * unexpected errors that left the session in a non-terminal state.
   *
   * It will reset session status to 'idle', mark any non-terminal messages
   * as 'cancelled', clear any pending interrupt data, and add a system
   * message indicating the reset.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session entity to reset.
   *
   * @return int
   *   The number of messages that were reset/cancelled.
   */
  public function resetSession(FlowDropSessionInterface $session): int {
    $sessionId = $session->id();
    if ($sessionId === NULL) {
      return 0;
    }

    $resetCount = 0;

    // Find and cancel any non-terminal messages.
    $query = $this->messageStorage->getQuery()
      ->condition('session_id', $sessionId)
      ->condition('status', [MessageStatus::PENDING, MessageStatus::PROCESSING, MessageStatus::AWAITING_INPUT], 'IN')
      ->accessCheck(TRUE);

    $messageIds = $query->execute();

    if (!empty($messageIds)) {
      $messages = $this->messageStorage->loadMultiple($messageIds);
      foreach ($messages as $message) {
        if ($message instanceof FlowDropSessionMessageInterface) {
          $message->setStatus(MessageStatus::CANCELLED)->save();
          $resetCount++;
        }
      }
    }

    // Reset session status to idle.
    $session->setStatus(SessionStatus::IDLE);
    $session->setCurrentPipelineId(NULL);
    $session->save();

    // Clear pending-interrupt slots on every pipeline bound to this
    // session. Pipelines own their own pending state now.
    $pipelineStorage = $this->entityTypeManager->getStorage('flowdrop_pipeline');
    $pipelineIds = $pipelineStorage->getQuery()
      ->condition('session_id', $sessionId)
      ->condition('pending_interrupt_id', NULL, 'IS NOT NULL')
      ->accessCheck(FALSE)
      ->execute();
    foreach ($pipelineStorage->loadMultiple($pipelineIds) as $pipeline) {
      if ($pipeline instanceof FlowDropPipelineInterface) {
        $pipeline->setPendingInterruptId(NULL);
        $pipeline->setPendingInterruptType(NULL);
        $pipeline->save();
      }
    }

    // Add system message indicating the reset.
    if ($resetCount > 0) {
      $this->addSystemMessage(
        $session,
        "Session reset: {$resetCount} pending message(s) cancelled. You can continue the conversation.",
        ['level' => 'info', 'type' => 'session_reset']
      );
    }
    else {
      $this->addSystemMessage(
        $session,
        "Session reset to idle state. You can continue the conversation.",
        ['level' => 'info', 'type' => 'session_reset']
      );
    }

    $this->logger->info('Reset session @id, cancelled @count messages', [
      '@id' => $sessionId,
      '@count' => $resetCount,
    ]);

    return $resetCount;
  }

  /**
   * Collects the assistant message ids written back during an execution.
   *
   * No access check: callers are services whose boundary (HTTP endpoint,
   * drush, queue worker) already enforced access to the session, and the
   * write-backs were created by the execution service itself.
   *
   * @param string $sessionId
   *   The session entity id.
   * @param string $executionId
   *   The pipeline (execution) id stamped on the write-backs.
   *
   * @return array<int, string>
   *   Assistant message ids in conversation order.
   */
  public function getAssistantMessageIdsForExecution(string $sessionId, string $executionId): array {
    $ids = $this->messageStorage->getQuery()
      ->condition('session_id', $sessionId)
      ->condition('execution_id', $executionId)
      ->condition('role', MessageRole::ASSISTANT)
      ->accessCheck(FALSE)
      ->sort('sequence_number', 'ASC')
      ->execute();
    return array_map('strval', array_values($ids));
  }

  /**
   * Finds the first ChatInput node in a workflow.
   *
   * @param \Drupal\flowdrop_workflow\WorkflowDefinitionInterface $workflow
   *   The workflow entity.
   *
   * @return array<string, mixed>|null
   *   The ChatInput node array or NULL if not found.
   */
  public function findChatInputNode(WorkflowDefinitionInterface $workflow): ?array {
    $nodes = $workflow->getNodes();
    $chatInputNodes = [];

    $patterns = ['chat_input.', 'text_input.'];

    foreach ($nodes as $node) {
      if (!is_array($node) || !isset($node['id'])) {
        continue;
      }

      // Check against node instance ID (e.g., "chat_input.1") not type ID.
      // The trailing dot in patterns prevents false matches
      // like "chat_input_custom.1".
      $nodeInstanceId = strtolower((string) $node['id']);

      foreach ($patterns as $pattern) {
        if (str_contains($nodeInstanceId, $pattern)) {
          $chatInputNodes[] = $node;
          break;
        }
      }
    }

    // Note: Session override would need to be passed separately if needed.
    // For now, we use the first match.
    // Return first match (warn if multiple found).
    if (!empty($chatInputNodes)) {
      if (count($chatInputNodes) > 1) {
        $this->logger->warning('Multiple ChatInput nodes found in workflow @id, using first', [
          '@id' => $workflow->id(),
        ]);
      }
      return $chatInputNodes[0];
    }

    return NULL;
  }

  /**
   * Formats a session entity for API response.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session entity.
   *
   * @return array<string, mixed>
   *   Formatted session data.
   */
  public function formatSessionForApi(FlowDropSessionInterface $session): array {
    $owner = $session->getOwner();

    return [
      'id' => $session->uuid(),
      'workflowId' => $session->getWorkflowId(),
      'name' => $session->getDisplayLabel(),
      'status' => $session->getStatus(),
      'createdAt' => date('c', (int) $session->get('created')->value),
      'updatedAt' => date('c', (int) $session->getChangedTime()),
      'metadata' => $session->getMetadata(),
      'executions' => $this->getExecutionsForSession($session),
      'owner' => $owner !== NULL ? [
        'id' => $owner->id(),
        'name' => $owner->getDisplayName(),
      ] : NULL,
    ];
  }

  /**
   * Derives the ordered execution list for a session from its user messages.
   *
   * User messages that have an execution_id carry the pipeline run's status
   * (completed or failed) after execution finishes. This lets the frontend
   * show correct "Run #N completed/failed" separators when loading historical
   * sessions without any extra storage.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session entity.
   *
   * @return array<int, array{id: string, startedAt: string, status: string}>
   *   Executions in chronological order, each with id, startedAt, and status.
   */
  protected function getExecutionsForSession(FlowDropSessionInterface $session): array {
    $session_id = $session->id();
    if ($session_id === NULL) {
      return [];
    }

    $rows = $this->database->select('flowdrop_session_message', 'm')
      ->fields('m', ['execution_id', 'timestamp', 'status'])
      ->condition('m.session_id', (int) $session_id)
      ->condition('m.role', MessageRole::USER)
      ->isNotNull('m.execution_id')
      ->orderBy('m.id', 'ASC')
      ->execute()
      ->fetchAll();

    $executions = [];
    foreach ($rows as $row) {
      $status = match ($row->status) {
        MessageStatus::COMPLETED => 'completed',
        MessageStatus::FAILED => 'failed',
        default => 'running',
      };
      $executions[] = [
        'id' => $row->execution_id,
        'startedAt' => date('c', (int) $row->timestamp),
        'status' => $status,
      ];
    }

    return $executions;
  }

  /**
   * Formats a list of message entities for API response.
   *
   * Resolves pipeline lineage (root / parent) for the whole page in one
   * query per session rather than per message, then formats each message
   * with the resolved map. Prefer this over looping formatMessageForApi()
   * so the lineage lookup stays batched.
   *
   * @param array<int, \Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface> $messages
   *   The message entities (typically one session's page).
   *
   * @return array<int, array<string, mixed>>
   *   Formatted message data, in input order.
   */
  public function formatMessagesForApi(array $messages): array {
    if ($messages === []) {
      return [];
    }

    // Group execution ids by session so each resolver lookup is
    // session-scoped (guards an external execution_id from colliding with an
    // unrelated pipeline id).
    $idsBySession = [];
    foreach ($messages as $message) {
      $executionId = $message->getExecutionId();
      $sessionId = $message->getSessionId();
      if ($executionId === NULL || $sessionId === 0) {
        continue;
      }
      $idsBySession[$sessionId][] = $executionId;
    }

    // Pipeline ids are globally unique, so merging per-session maps is safe.
    $lineage = [];
    foreach ($idsBySession as $sessionId => $executionIds) {
      $lineage += $this->lineageResolver->resolveForSession($executionIds, $sessionId);
    }

    $data = [];
    foreach ($messages as $message) {
      $data[] = $this->formatMessageForApi($message, $lineage);
    }
    return $data;
  }

  /**
   * Formats a message entity for API response.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface $message
   *   The message entity.
   * @param array<string, array{root: string, parent: string|null}> $lineage
   *   Optional pre-resolved lineage map keyed by pipeline id, from
   *   formatMessagesForApi(). When the message's execution_id is present, the
   *   rootPipelineId / parentPipelineId wire fields are added. Pass empty to
   *   skip lineage (single-message callers may resolve one id themselves).
   *
   * @return array<string, mixed>
   *   Formatted message data.
   */
  public function formatMessageForApi(FlowDropSessionMessageInterface $message, array $lineage = []): array {
    $metadata = $message->getMessageMetadata();

    // Lift server-emitted annotations (hierarchy/tags/display) from metadata
    // to top-level wire fields per the @flowdrop/flowdrop 1.13.0 contract.
    // Malformed shapes are dropped silently so a bad emitter cannot break the
    // playground response for the rest of the conversation.
    $hierarchy = $this->extractHierarchy($metadata);
    $tags = $this->extractTags($metadata);
    $display = $this->extractDisplay($metadata);
    unset(
      $metadata[MessageAnnotation::METADATA_KEY_HIERARCHY],
      $metadata[MessageAnnotation::METADATA_KEY_TAGS],
      $metadata[MessageAnnotation::METADATA_KEY_DISPLAY],
    );

    $data = [
      'id' => $message->uuid(),
      'sessionId' => $message->getSession()?->uuid(),
      'role' => $message->getRole(),
      'content' => $message->getContent(),
      'timestamp' => date('c', $message->getTimestamp()),
      'status' => $message->getStatus(),
      'sequenceNumber' => $message->getSequenceNumber(),
      'nodeId' => $message->getNodeId(),
      'metadata' => $metadata,
    ];

    if ($hierarchy !== NULL) {
      $data['hierarchy'] = $hierarchy;
    }
    if ($tags !== NULL) {
      $data['tags'] = $tags;
    }
    if ($display !== NULL) {
      $data['display'] = $display;
    }

    // Add optional linking fields.
    $parentMessageId = $message->getParentMessageId();
    if ($parentMessageId !== NULL) {
      $parentMessage = $this->messageStorage->load($parentMessageId);
      if ($parentMessage instanceof FlowDropSessionMessageInterface) {
        $data['parentMessageId'] = $parentMessage->uuid();
      }
    }

    $executionId = $message->getExecutionId();
    if ($executionId !== NULL) {
      $data['executionId'] = $executionId;
      // Root/parent pipeline of the run that produced this message. Resolved
      // via the pipeline (not stored on the message); absent when the
      // execution_id is not a pipeline in this session.
      if (isset($lineage[$executionId])) {
        $data['rootPipelineId'] = $lineage[$executionId]['root'];
        $data['parentPipelineId'] = $lineage[$executionId]['parent'];
      }
    }

    return $data;
  }

  /**
   * Extracts a validated hierarchy list from message metadata.
   *
   * @param array<string, mixed> $metadata
   *   The decoded message metadata.
   *
   * @return array<int, array{id: string, label: string, icon?: string}>|null
   *   The cleaned hierarchy items, or NULL if no valid items were present.
   */
  private function extractHierarchy(array $metadata): ?array {
    $raw = $metadata[MessageAnnotation::METADATA_KEY_HIERARCHY] ?? NULL;
    if (!is_array($raw) || $raw === []) {
      return NULL;
    }

    $cleaned = [];
    foreach ($raw as $item) {
      if (!is_array($item)) {
        continue;
      }
      $id = $item['id'] ?? NULL;
      $label = $item['label'] ?? NULL;
      if (!is_string($id) || $id === '' || !is_string($label) || $label === '') {
        continue;
      }
      $entry = ['id' => $id, 'label' => $label];
      $icon = $item['icon'] ?? NULL;
      if (is_string($icon) && $icon !== '') {
        $entry['icon'] = $icon;
      }
      $cleaned[] = $entry;
    }

    return $cleaned !== [] ? $cleaned : NULL;
  }

  /**
   * Extracts a validated tag list from message metadata.
   *
   * @param array<string, mixed> $metadata
   *   The decoded message metadata.
   *
   * @return array<int, array{id: string, label: string, icon?: string, color?: string, variant?: string, type?: string}>|null
   *   The cleaned tags, or NULL if no valid tags were present.
   */
  private function extractTags(array $metadata): ?array {
    $raw = $metadata[MessageAnnotation::METADATA_KEY_TAGS] ?? NULL;
    if (!is_array($raw) || $raw === []) {
      return NULL;
    }

    $cleaned = [];
    foreach ($raw as $item) {
      if (!is_array($item)) {
        continue;
      }
      $id = $item['id'] ?? NULL;
      $label = $item['label'] ?? NULL;
      if (!is_string($id) || $id === '' || !is_string($label) || $label === '') {
        continue;
      }
      $entry = ['id' => $id, 'label' => $label];

      $icon = $item['icon'] ?? NULL;
      if (is_string($icon) && $icon !== '') {
        $entry['icon'] = $icon;
      }

      $color = $item['color'] ?? NULL;
      if (is_string($color) && in_array($color, MessageAnnotation::COLORS, TRUE)) {
        $entry['color'] = $color;
      }

      $variant = $item['variant'] ?? NULL;
      if (is_string($variant) && in_array($variant, MessageAnnotation::VARIANTS, TRUE)) {
        $entry['variant'] = $variant;
      }

      $type = $item['type'] ?? NULL;
      if (is_string($type) && $type !== '') {
        $entry['type'] = $type;
      }

      $cleaned[] = $entry;
    }

    return $cleaned !== [] ? $cleaned : NULL;
  }

  /**
   * Extracts a validated display hint from message metadata.
   *
   * @param array<string, mixed> $metadata
   *   The decoded message metadata.
   *
   * @return string|null
   *   The display value, or NULL if absent or not in the allowed enum.
   */
  private function extractDisplay(array $metadata): ?string {
    $display = $metadata[MessageAnnotation::METADATA_KEY_DISPLAY] ?? NULL;
    if (is_string($display) && in_array($display, MessageAnnotation::DISPLAYS, TRUE)) {
      return $display;
    }
    return NULL;
  }

}
