<?php

declare(strict_types=1);

namespace Drupal\flowdrop_memory\Plugin\FlowDropNodeProcessor;

use Drupal\Core\StringTranslation\TranslatableMarkup;
use Drupal\flowdrop\Attribute\FlowDropNodeProcessor;
use Drupal\flowdrop\DTO\ParameterBagInterface;
use Drupal\flowdrop\DTO\ValidationError;
use Drupal\flowdrop\DTO\ValidationResult;
use Drupal\flowdrop\Exception\InvalidNodeConfigurationException;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\AbstractFlowDropNodeProcessor;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\ExecutionContextAwareInterface;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\HasSideEffectsInterface;
use Drupal\flowdrop_memory\Service\MemoryManager;
use Symfony\Component\DependencyInjection\ContainerInterface;

/**
 * Conversation buffer node processor.
 *
 * Appends one or more messages to a stored conversation history list and writes
 * it back atomically. Scope ID is auto-resolved from the execution context.
 *
 * Each message is normalized to the Message-DTO array shape
 * `{role, content, metadata, timestamp}` (matching
 * \Drupal\flowdrop_orchestration\DTO\Snapshot\Message::toArray()), so a buffer
 * round-trips with the native runtime. Agent-loop conversations are supported:
 * an assistant turn's `tool_calls` and a `tool`-role turn's `tool_call_id`
 * (accepted at the top level or under `metadata`) are preserved under
 * `metadata` — a provider replaying the history sees paired tool_use and
 * tool_result turns instead of the stripped `{role, content}` it used to keep.
 *
 * Two ways to append:
 * - the `role` + `content` inputs (a single simple message), or
 * - the `message` input: a full message object or an array of them. When
 *   `message` is connected, `role`/`content` are ignored.
 *
 * Covers the LangChain ConversationBufferMemory and
 * ConversationBufferWindowMemory patterns in one node. Set max_messages > 0 to
 * enable a sliding window; the oldest messages are discarded first, but the
 * window never splits a `tool_use` from its `tool_result` (it cuts on a user
 * turn at/after the boundary, or keeps the buffer whole if none exists).
 *
 * @see \Drupal\flowdrop_memory\Plugin\FlowDropNodeProcessor\MemoryDelete
 */
#[FlowDropNodeProcessor(
  id: 'conversation_buffer',
  label: new TranslatableMarkup('Conversation Buffer'),
  description: 'Append a message to the conversation buffer. Scope ID is automatically derived from the execution context.',
  version: '1.0.0',
)]
class ConversationBuffer extends AbstractFlowDropNodeProcessor implements ExecutionContextAwareInterface, HasSideEffectsInterface {

  use ScopeResolutionTrait;

  /**
   * {@inheritdoc}
   */
  public function __construct(
    array $configuration,
    string $plugin_id,
    mixed $plugin_definition,
    private readonly MemoryManager $memoryManager,
  ) {
    parent::__construct($configuration, $plugin_id, $plugin_definition);
  }

  /**
   * {@inheritdoc}
   */
  public static function create(ContainerInterface $container, array $configuration, $plugin_id, $plugin_definition): static {
    return new static(
      $configuration,
      $plugin_id,
      $plugin_definition,
      $container->get('flowdrop_memory.manager'),
    );
  }

  /**
   * {@inheritdoc}
   */
  public function validateParams(array $params): ValidationResult {
    // When a full message (or array of messages) is provided, role/content are
    // ignored, so they are not required. Message-shape validation (role set,
    // tool turns carry a tool_call_id) happens loudly at normalization time.
    if (self::hasMessageInput($params['message'] ?? NULL)) {
      return ValidationResult::success();
    }

    $errors = [];
    if (empty($params['role'])) {
      $errors[] = new ValidationError('role', 'Role is required when no message is provided.');
    }
    if (!array_key_exists('content', $params) || $params['content'] === NULL) {
      $errors[] = new ValidationError('content', 'Content is required when no message is provided.');
    }
    return empty($errors) ? ValidationResult::success() : ValidationResult::failure($errors);
  }

  /**
   * {@inheritdoc}
   */
  public function process(ParameterBagInterface $params): array {
    $scope = $params->getString('scope', 'session');
    $key = $params->getString('key', 'chat_history');
    $maxMessages = $params->getInt('max_messages', 0);
    $backend = $params->getString('backend', 'static');

    $resolvedScopeId = $this->resolveScopeId($scope);

    $messages = $this->memoryManager->get($scope, $resolvedScopeId, $key, [], $backend);
    if (!is_array($messages)) {
      $messages = [];
    }

    foreach ($this->incomingMessages($params) as $raw) {
      $normalized = self::normalizeMessage($raw);
      // A NULL result is an intentional drop (empty content, no tool call).
      if ($normalized !== NULL) {
        $messages[] = $normalized;
      }
    }

    $messages = self::applyWindow(array_values($messages), $maxMessages);

    $this->memoryManager->set($scope, $resolvedScopeId, $key, $messages, NULL, $backend);

    return [
      'messages'          => $messages,
      'count'             => count($messages),
      'key'               => $key,
      'scope'             => $scope,
      'resolved_scope_id' => $resolvedScopeId,
    ];
  }

  /**
   * Resolves the raw messages to append from the node's inputs.
   *
   * The `message` input (a single message object or a list of them) takes
   * precedence; when it is absent the simple `role` + `content` inputs form one
   * message. A `message` provided as a JSON string is decoded.
   *
   * @param \Drupal\flowdrop\DTO\ParameterBagInterface $params
   *   The resolved node parameters.
   *
   * @return array<int, array<string, mixed>>
   *   A list of raw (un-normalized) message arrays.
   */
  private function incomingMessages(ParameterBagInterface $params): array {
    $message = $params->get('message');
    if (self::hasMessageInput($message)) {
      if (is_string($message)) {
        $decoded = json_decode($message, TRUE);
        $message = is_array($decoded) ? $decoded : NULL;
      }
      if (!is_array($message)) {
        return [];
      }
      // A single message object carries a 'role'; otherwise treat a list as a
      // batch of messages and ignore any non-array elements.
      if (array_key_exists('role', $message)) {
        return [$message];
      }
      if (array_is_list($message)) {
        return array_values(array_filter($message, 'is_array'));
      }
      return [];
    }

    return [
      [
        'role'    => $params->getString('role'),
        'content' => $params->get('content'),
      ],
    ];
  }

  /**
   * Whether a usable `message` input value is present.
   *
   * @param mixed $message
   *   The raw `message` input.
   *
   * @return bool
   *   TRUE when the value is a non-empty array or string.
   */
  private static function hasMessageInput(mixed $message): bool {
    if (is_array($message)) {
      return $message !== [];
    }
    return is_string($message) && $message !== '';
  }

  /**
   * Normalizes one raw message into the Message-DTO array shape.
   *
   * Provider-flat `tool_calls` / `tool_call_id` (or the same under `metadata`)
   * are lifted into `metadata`; any other non-reserved top-level key is also
   * preserved under `metadata`.
   *
   * @param array<string, mixed> $raw
   *   A raw message array.
   *
   * @return array<string, mixed>|null
   *   The normalized `{role, content, metadata, timestamp}` message, or NULL if
   *   it should be dropped (empty content that is not a silent tool call).
   *
   * @throws \Drupal\flowdrop\Exception\InvalidNodeConfigurationException
   *   When the message has no role, or a `tool` turn lacks a `tool_call_id` —
   *   providers reject such a history, so fail here, not at the provider.
   */
  private static function normalizeMessage(array $raw): ?array {
    $role = $raw['role'] ?? NULL;
    if (!is_string($role) || $role === '') {
      throw new InvalidNodeConfigurationException('Each conversation message requires a non-empty "role".');
    }

    $metadata = (isset($raw['metadata']) && is_array($raw['metadata'])) ? $raw['metadata'] : [];

    $toolCalls = $raw['tool_calls'] ?? $metadata['tool_calls'] ?? NULL;
    if ($toolCalls !== NULL) {
      $metadata['tool_calls'] = $toolCalls;
    }
    $toolCallId = $raw['tool_call_id'] ?? $metadata['tool_call_id'] ?? NULL;
    if ($toolCallId !== NULL) {
      $metadata['tool_call_id'] = $toolCallId;
    }

    // Preserve any other non-reserved top-level keys under metadata.
    $reserved = ['role', 'content', 'timestamp', 'metadata', 'tool_calls', 'tool_call_id'];
    foreach ($raw as $name => $value) {
      if (!in_array($name, $reserved, TRUE)) {
        $metadata[$name] = $value;
      }
    }

    if ($role === 'tool' && empty($metadata['tool_call_id'])) {
      throw new InvalidNodeConfigurationException('A "tool" role message requires a "tool_call_id".');
    }

    $content = $raw['content'] ?? '';

    // Empty-content rules: an assistant turn that only calls tools is kept
    // (models call tools silently); any other empty message is dropped.
    // Content can never be NULL here — it defaults to '' above.
    $isEmpty = $content === '';
    if ($isEmpty && !($role === 'assistant' && !empty($metadata['tool_calls']))) {
      return NULL;
    }

    return [
      'role'      => $role,
      'content'   => $content,
      'metadata'  => $metadata,
      'timestamp' => $raw['timestamp'] ?? time(),
    ];
  }

  /**
   * Applies the sliding window without splitting a tool_use/tool_result pair.
   *
   * A tool-free buffer trims with a plain tail slice (unchanged behavior).
   * Once any message carries tool metadata, the cut moves to the first
   * `user` turn at or after the naive boundary; if there is no such turn,
   * the whole buffer is kept rather than orphan a tool_use or tool_result.
   *
   * @param array<int, array<string, mixed>> $messages
   *   The full (list-indexed) message buffer.
   * @param int $maxMessages
   *   The window size; 0 or negative means unlimited.
   *
   * @return array<int, array<string, mixed>>
   *   The windowed buffer.
   */
  private static function applyWindow(array $messages, int $maxMessages): array {
    if ($maxMessages <= 0 || count($messages) <= $maxMessages) {
      return $messages;
    }

    if (!self::hasToolMetadata($messages)) {
      return array_slice($messages, -$maxMessages);
    }

    $naiveStart = count($messages) - $maxMessages;
    for ($i = $naiveStart; $i < count($messages); $i++) {
      if (($messages[$i]['role'] ?? NULL) === 'user') {
        return array_slice($messages, $i);
      }
    }

    // No user boundary past the cut point — keep the buffer whole.
    return $messages;
  }

  /**
   * Whether any message in the buffer carries tool-calling metadata.
   *
   * @param array<int, array<string, mixed>> $messages
   *   The message buffer.
   *
   * @return bool
   *   TRUE if any message is a tool turn or carries tool_calls/tool_call_id.
   */
  private static function hasToolMetadata(array $messages): bool {
    foreach ($messages as $message) {
      if (($message['role'] ?? NULL) === 'tool') {
        return TRUE;
      }
      $metadata = $message['metadata'] ?? [];
      if (is_array($metadata) && (!empty($metadata['tool_calls']) || !empty($metadata['tool_call_id']))) {
        return TRUE;
      }
    }
    return FALSE;
  }

  /**
   * {@inheritdoc}
   */
  public function getParameterSchema(): array {
    return [
      'type' => 'object',
      'properties' => [
        'scope' => [
          'type'        => 'string',
          'title'       => 'Scope',
          'description' => 'The memory scope.',
          'enum'        => ['global', 'workflow', 'pipeline', 'execution', 'session'],
          'default'     => 'session',
        ],
        'key' => [
          'type'        => 'string',
          'title'       => 'Key',
          'description' => 'Memory key for the conversation buffer.',
          'default'     => 'chat_history',
        ],
        'message' => [
          'type'        => 'mixed',
          'title'       => 'Message(s)',
          'description' => 'A full message object {role, content, tool_calls?, tool_call_id?, metadata?} or an array of them. When provided, the role/content inputs are ignored. tool_calls/tool_call_id may be given at the top level or under metadata.',
        ],
        'role' => [
          'type'        => 'string',
          'title'       => 'Role',
          'description' => 'The message role. Ignored when the message input is connected.',
          'enum'        => ['user', 'assistant', 'system', 'tool'],
        ],
        'content' => [
          'type'        => 'mixed',
          'title'       => 'Content',
          'description' => 'The message content. Ignored when the message input is connected.',
        ],
        'max_messages' => [
          'type'        => 'integer',
          'title'       => 'Max Messages',
          'description' => 'Sliding window size. 0 means unlimited (buffer memory).',
          'default'     => 0,
          'minimum'     => 0,
        ],
        'backend' => [
          'type'        => 'string',
          'title'       => 'Memory Backend',
          'description' => 'The storage backend to use.',
          'enum'        => ['static', 'cached', 'entity'],
          'default'     => 'static',
        ],
      ],
      // role/content are conditionally required (only when no message input);
      // enforced in validateParams(), not declared schema-level.
      'required' => ['backend'],
    ];
  }

  /**
   * {@inheritdoc}
   */
  public function getOutputSchema(): array {
    return [
      'type' => 'object',
      'properties' => [
        'messages' => [
          'type'        => 'array',
          'description' => 'Full conversation buffer as [{role, content, metadata, timestamp}]; tool_calls/tool_call_id live under metadata.',
        ],
        'count' => [
          'type'        => 'integer',
          'description' => 'Number of messages currently in the buffer.',
        ],
        'key' => [
          'type'        => 'string',
          'description' => 'The key used to store the buffer.',
        ],
        'scope' => [
          'type'        => 'string',
          'description' => 'The scope used.',
        ],
        'resolved_scope_id' => [
          'type'        => 'string',
          'description' => 'The scope ID that was actually used (after auto-resolution).',
        ],
      ],
    ];
  }

}
