<?php

namespace Drupal\acquia_contenthub_subscriber\Commands;

use Acquia\ContentHubClient\ContentHubClient;
use Consolidation\AnnotatedCommand\CommandError;
use Drupal\acquia_contenthub\Client\ClientFactory;
use Drupal\acquia_contenthub\Commands\AcquiaContentHubQueueCommands;
use Drupal\acquia_contenthub\Libs\Common\EnsureContentHubClientTrait;
use Drupal\acquia_contenthub\Libs\Logging\ContentHubEventLogger;
use Drupal\acquia_contenthub\Libs\ServiceQueue\ServiceQueueHandler;
use Drupal\acquia_contenthub\Libs\Traits\ResponseCheckerTrait;
use Drupal\acquia_contenthub\Settings\ConnectionDetailsInterface;
use Drupal\acquia_contenthub_subscriber\ContentHubImportQueue;
use Drupal\Core\Queue\QueueFactory;
use Drupal\Core\Queue\QueueWorkerManager;
use Psr\Log\LoggerInterface;
use Symfony\Component\Console\Helper\ProgressBar;

/**
 * Import Queue Run Command.
 */
class AcquiaContentHubSubscriberImportQueueCommands extends AcquiaContentHubQueueCommands {

  use EnsureContentHubClientTrait;
  use ResponseCheckerTrait;

  /**
   * Queue Name.
   */
  public const QUEUE_NAME = 'acquia_contenthub_subscriber_import';

  /**
   * Queue process.
   */
  public const QUEUE_PROCESS = 'Import queue';

  /**
   * The connection details service.
   *
   * @var \Drupal\acquia_contenthub\Settings\ConnectionDetailsInterface
   */
  protected ConnectionDetailsInterface $connectionDetails;

  /**
   * The service queue handler.
   *
   * @var \Drupal\acquia_contenthub\Libs\ServiceQueue\ServiceQueueHandler
   */
  protected ServiceQueueHandler $serviceQueueHandler;

  /**
   * Constructs AcquiaContentHubSubscriberImportQueueCommands object.
   *
   * @param \Drupal\Core\Queue\QueueWorkerManager $queue_worker
   *   Queue Worker Manager.
   * @param \Drupal\Core\Queue\QueueFactory $queue_factory
   *   Queue Factory.
   * @param \Drupal\acquia_contenthub\Libs\Logging\ContentHubEventLogger $event_logger
   *   CH Event logger.
   * @param \Psr\Log\LoggerInterface $logger
   *   Logger.
   * @param \Drupal\acquia_contenthub\Client\ClientFactory $client_factory
   *   The client factory service.
   * @param \Drupal\acquia_contenthub\Settings\ConnectionDetailsInterface $connection_details
   *   The connection details service.
   * @param \Drupal\acquia_contenthub\Libs\ServiceQueue\ServiceQueueHandler $service_queue_handler
   *   The service queue handler.
   */
  public function __construct(QueueWorkerManager $queue_worker, QueueFactory $queue_factory, ContentHubEventLogger $event_logger, LoggerInterface $logger, ClientFactory $client_factory, ConnectionDetailsInterface $connection_details, ServiceQueueHandler $service_queue_handler) {
    parent::__construct($queue_worker, $queue_factory, $event_logger, $logger);
    $this->clientFactory = $client_factory;
    $this->connectionDetails = $connection_details;
    $this->serviceQueueHandler = $service_queue_handler;
  }

  /**
   * Runs Content Hub subscriber import queue and service queue.
   *
   * @option time-limit
   *     The maximum number of seconds allowed to run the queue.
   * @default time-limit
   *
   * @option items-limit
   *      The maximum number of items allowed to run the queue.
   * @default items-limit
   *
   * @option service-queue-item-state
   *      Filter service queue items by state. Eg: 'queued', 'failed'.
   * @default service-queue-item-state queued
   *
   * @usage drush acquia:contenthub-import-queue-run --uri=https://site_uri
   *   | Runs the queue.
   * @usage drush acquia:contenthub-import-queue-run --service-queue-item-state=queued
   *   | Runs the queue with service queue item state filter.
   * @usage drush acquia:contenthub-import-queue-run --service-queue-item-state=queued,failed
   *   | Runs the queue processing both queued and failed items.
   * @usage drush acquia:contenthub-import-queue-run --items-limit=100 --time-limit=300
   *   | Runs the queue with a limit of 100 items and 5 minutes time limit.
   *
   * @command acquia:contenthub-import-queue-run
   * @aliases ach-import
   *
   * @return int
   *   Exit code: 0 for success, non-zero for errors.
   *
   * @throws \Drupal\Component\Plugin\Exception\PluginException
   */
  public function run(): int {
    $this->runQueue();
    if ($this->connectionDetails->getSyndicationMode() === ConnectionDetailsInterface::SYNDICATION_MODE_PULL) {
      return $this->handlePullSyndicationQueue();
    }
    return self::EXIT_SUCCESS;
  }

  /**
   * Validates the service-queue-item-state option.
   *
   * @hook validate acquia:contenthub-import-queue-run
   */
  public function validateServiceQueueItemState(): ?CommandError {
    $allowed_states = ['queued', 'failed'];
    $service_queue_item_states = $this->getServiceQueueFilters();

    if (empty($service_queue_item_states)) {
      return NULL;
    }
    $invalid_states = array_diff($service_queue_item_states, $allowed_states);
    if (!empty($invalid_states)) {
      return new CommandError(sprintf(
        'Invalid service-queue-item-state values: %s. Allowed values are: %s',
        implode(', ', $invalid_states),
        implode(', ', $allowed_states)
      ), self::EXIT_FAILURE);
    }

    return NULL;
  }

  /**
   * Handles pull syndication queue processing.
   *
   * @return int
   *   Exit code: 0 for success, non-zero for errors.
   */
  protected function handlePullSyndicationQueue(): int {
    $queue_filters = $this->getServiceQueueFilters();
    $time_limit = (int) $this->input()->getOption('time-limit');
    $items_limit = (int) $this->input()->getOption('items-limit');
    $start_time = time();
    $end_time = $time_limit ? $start_time + $time_limit : 0;

    return $this->processServiceQueueItems($queue_filters, $end_time, $items_limit);
  }

  /**
   * Gets service queue filters from the command option.
   *
   * @return array
   *   Array of queue filters.
   */
  protected function getServiceQueueFilters(): array {
    $queue_filters = [];
    $service_queue_item_state = $this->input()->getOption('service-queue-item-state');
    if ($service_queue_item_state) {
      $queue_filters = array_map('trim', explode(',', $service_queue_item_state));
    }
    return $queue_filters;
  }

  /**
   * Processes service queue items in batches.
   *
   * @param array $queue_filters
   *   Queue filters to apply.
   * @param int $end_time
   *   End time timestamp (0 if no time limit).
   * @param int $items_limit
   *   Items limit.
   *
   * @return int
   *   Exit code: 0 for success, non-zero for errors.
   */
  protected function processServiceQueueItems(array $queue_filters, int $end_time, int $items_limit): int {
    $client = $this->getClient();
    $total_processed = 0;
    $symfony_progressbar = $this->bootstrapProgressBar($queue_filters, $items_limit, $client);
    $symfony_progressbar->start();

    try {
      do {
        if ($end_time && time() >= $end_time) {
          $this->logQueueProcess('Time limit reached, stopping queue processing.', self::QUEUE_PROCESS);
          break;
        }
        if ($items_limit && $total_processed >= $items_limit) {
          $this->logQueueProcess('Items limit reached, stopping queue processing.', self::QUEUE_PROCESS);
          break;
        }
        $batch_limit = ServiceQueueHandler::LIMIT;
        if ($items_limit) {
          $remaining_items = $items_limit - $total_processed;
          $batch_limit = min($batch_limit, $remaining_items);
        }

        $response = $client->receiveQueueItems($batch_limit, ServiceQueueHandler::VISIBILITY_TIMEOUT, $queue_filters);
        if (!$this->isResponseSuccessful($response)) {
          $this->logQueueProcess(sprintf(
            'Error communicating with Content Hub while fetching queue items: %s', ($response['error']['message'] ?? 'Unknown')
          ), self::QUEUE_PROCESS);
          $symfony_progressbar->finish();

          return self::EXIT_FAILURE;
        }

        $queue_items = $response['data'] ?? [];
        if (empty($queue_items)) {
          break;
        }

        $this->serviceQueueHandler->processQueueItems($queue_items);
        $total_processed += count($queue_items);
        $symfony_progressbar->advance(count($queue_items));
      } while (count($response['data']) === ServiceQueueHandler::LIMIT);

      $symfony_progressbar->finish();
      $elapsed = time() - $symfony_progressbar->getStartTime();
      if ($total_processed > 0) {
        $this->logQueueProcess("Successfully processed {$total_processed} service queue items in {$elapsed} seconds", self::QUEUE_PROCESS);
        $this->logger()->success(dt("Successfully processed {$total_processed} service queue items in {$elapsed} seconds"));
      }
      else {
        $this->logger()->success(dt('No entities found in Content Hub Service queue.'));
      }

      return self::EXIT_SUCCESS;
    }
    catch (\Exception $e) {
      $this->logQueueProcess("Error processing service queue items: {$e->getMessage()}", self::QUEUE_PROCESS);
      $symfony_progressbar->finish();

      return self::EXIT_FAILURE;
    }
  }

  /**
   * Bootstraps symfony ProgressBar.
   *
   * Sets the source count to the total number of available queue items,
   * capped at the items limit if the total exceeds it.
   *
   * @param array $states
   *   The states to filter by.
   * @param int $items_limit
   *   The limit of how many items should be processed at once.
   * @param \Acquia\ContentHubClient\ContentHubClient $client
   *   The Content Hub client instance.
   *
   * @return \Symfony\Component\Console\Helper\ProgressBar
   *   The bootstrapped progress bar with the appropriate source number.
   *
   * @throws \Exception
   */
  protected function bootstrapProgressBar(array $states, int $items_limit, ContentHubClient $client): ProgressBar {
    $this->logQueueProcess('Starting to process service queue items in batches.', self::QUEUE_PROCESS);
    $source_count = $client->getAllQueueItems([
      'state' => implode(',', $states),
    ])['total'] ?? 0;

    if ($source_count > $items_limit) {
      $source_count = $items_limit;
    }

    return $this->initializeProgress($source_count);
  }

  /**
   * {@inheritDoc}
   */
  protected function getItemProcessingMessage($data): string {
    $message = '';
    if (isset($data->uuids)) {
      $message = "Processing entities: ($data->uuids) from import queue.";
    }
    return $message;
  }

  /**
   * {@inheritDoc}
   */
  protected function getQueueName(): string {
    return ContentHubImportQueue::QUEUE_NAME;
  }

  /**
   * {@inheritDoc}
   */
  protected function getQueueProcessName(): string {
    return self::QUEUE_PROCESS;
  }

}
