<?php

declare(strict_types=1);

namespace Drupal\ffmpeg_media\Commands;

use Drupal\Core\Config\ConfigFactoryInterface;
use Drupal\Core\Database\Connection;
use Drupal\Core\Database\DatabaseException;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Queue\QueueFactory;
use Drupal\ffmpeg_media\Exception\ConcurrentLimitException;
use Drupal\ffmpeg_media\Exception\FFmpegException;
use Drupal\ffmpeg_media\Service\DeadLetterStorageInterface;
use Drupal\ffmpeg_media\Service\LockManagerInterface;
use Drupal\ffmpeg_media\Service\ProgressTrackerInterface;
use Drupal\ffmpeg_media\Service\QueueManagerInterface;
use Drupal\ffmpeg_media\Service\QueueSchedulingServiceInterface;
use Drupal\ffmpeg_media\Service\RetryManagerInterface;
use Drupal\ffmpeg_media\Service\VideoConverterInterface;
use Drupal\ffmpeg_media\Utility\QueueItemKey;
use Drupal\media\MediaInterface;
use Drush\Attributes as CLI;
use Drush\Commands\DrushCommands;

/**
 * Drush commands for FFmpeg Media queue processing and management.
 */
final class QueueCommands extends DrushCommands {

  /**
   * Constructs a QueueCommands instance.
   *
   * @param \Drupal\ffmpeg_media\Service\VideoConverterInterface $videoConverter
   *   Video converter service.
   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entityTypeManager
   *   Entity type manager.
   * @param \Drupal\Core\Config\ConfigFactoryInterface $configFactory
   *   Config factory.
   * @param \Drupal\Core\Queue\QueueFactory $queueFactory
   *   Queue factory.
   * @param \Drupal\Core\Database\Connection $database
   *   Database connection.
   * @param \Drupal\ffmpeg_media\Service\QueueSchedulingServiceInterface $queueScheduling
   *   Queue scheduling service.
   * @param \Drupal\ffmpeg_media\Service\LockManagerInterface|null $lockManager
   *   Lock manager for concurrent conversion prevention.
   * @param \Drupal\ffmpeg_media\Service\RetryManagerInterface|null $retryManager
   *   Retry manager for exponential backoff.
   * @param \Drupal\ffmpeg_media\Service\ProgressTrackerInterface|null $progressTracker
   *   Progress tracker for failure recording.
   * @param \Drupal\ffmpeg_media\Service\DeadLetterStorageInterface|null $deadLetterStorage
   *   Dead-letter storage for permanently failed jobs.
   * @param \Drupal\ffmpeg_media\Service\QueueManagerInterface|null $queueManager
   *   Queue manager for dedup tracking and stale-item detection.
   */
  public function __construct(
    protected readonly VideoConverterInterface $videoConverter,
    protected readonly EntityTypeManagerInterface $entityTypeManager,
    protected readonly ConfigFactoryInterface $configFactory,
    protected readonly QueueFactory $queueFactory,
    protected readonly Connection $database,
    protected readonly QueueSchedulingServiceInterface $queueScheduling,
    protected readonly ?LockManagerInterface $lockManager = NULL,
    protected readonly ?RetryManagerInterface $retryManager = NULL,
    protected readonly ?ProgressTrackerInterface $progressTracker = NULL,
    protected readonly ?DeadLetterStorageInterface $deadLetterStorage = NULL,
    protected readonly ?QueueManagerInterface $queueManager = NULL,
  ) {
    parent::__construct();
  }

  /**
   * Process videos from the conversion queue.
   *
   * @param array $options
   *   Command options.
   *
   * @phpstan-param array{workers: int, limit: int, queue: string} $options
   */
  #[CLI\Command(name: 'ffmpeg:process', aliases: ['ffp'])]
  #[CLI\Option(name: 'workers', description: 'Requested worker count (1-4). Values greater than 1 currently fall back to sequential processing for stable process isolation.')]
  #[CLI\Option(name: 'limit', description: 'Maximum number of videos to process (0 = no limit).')]
  #[CLI\Option(name: 'queue', description: 'Queue to process: normal, large, or all.')]
  #[CLI\Usage(name: 'ffmpeg:process', description: 'Process queued videos with default settings.')]
  #[CLI\Usage(name: 'ffmpeg:process --workers=2 --limit=10', description: 'Process up to 10 videos; worker values greater than 1 currently run sequentially.')]
  #[CLI\Usage(name: 'ffmpeg:process --queue=large', description: 'Only process the large file queue.')]
  public function processQueue(array $options = ['workers' => 1, 'limit' => 0, 'queue' => 'all']): void {
    $workers = min(4, max(1, (int) $options['workers']));
    $limit = (int) $options['limit'];
    $queueOption = $options['queue'] ?? 'all';

    if ($workers > 1) {
      $this->logger()?->warning('Parallel worker mode is disabled for stable queue processing. Falling back to a single sequential worker.');
      $workers = 1;
    }

    $queueMap = [
      'normal' => QueueManagerInterface::QUEUE_NAME,
      'large' => QueueManagerInterface::QUEUE_LARGE_NAME,
    ];

    $queuesToProcess = [];
    if ($queueOption === 'all') {
      $queuesToProcess = array_values($queueMap);
    }
    elseif (isset($queueMap[$queueOption])) {
      $queuesToProcess[] = $queueMap[$queueOption];
    }
    else {
      $this->logger()?->error('Invalid queue option: {queue}. Use normal, large, or all.', ['queue' => $queueOption]);
      return;
    }

    foreach ($queuesToProcess as $queueName) {
      $queue = $this->queueFactory->get($queueName);
      $totalItems = $queue->numberOfItems();

      if ($totalItems === 0) {
        $this->logger()?->notice('No items in {queue} queue to process.', ['queue' => $queueName]);
        continue;
      }

      $toProcess = $limit > 0 ? min($limit, $totalItems) : $totalItems;
      $this->io()->title("Processing {$toProcess} videos from {$queueName} with {$workers} worker(s)");
      $this->processSequential($toProcess, $queueName);
    }
  }

  /**
   * Clear all video conversion queues.
   *
   * @param array $options
   *   Command options.
   *
   * @phpstan-param array{'dry-run': bool} $options
   */
  #[CLI\Command(name: 'ffmpeg:queue-clear', aliases: ['ffqc'])]
  #[CLI\Option(name: 'dry-run', description: 'Show what would be cleared without actually clearing.')]
  #[CLI\Usage(name: 'ffmpeg:queue-clear', description: 'Clear all items from the conversion queues.')]
  #[CLI\Usage(name: 'ffmpeg:queue-clear --dry-run', description: 'Show what would be cleared.')]
  public function queueClear(array $options = ['dry-run' => FALSE]): void {
    $dryRun = (bool) $options['dry-run'];

    $normalQueue = $this->queueFactory->get(QueueManagerInterface::QUEUE_NAME);
    $normalCount = $normalQueue->numberOfItems();

    $largeQueue = $this->queueFactory->get(QueueManagerInterface::QUEUE_LARGE_NAME);
    $largeCount = $largeQueue->numberOfItems();

    if ($dryRun) {
      $this->io()->title('[DRY RUN] Queue Clear Preview');
      $this->io()->listing([
        "Normal queue: {$normalCount} item(s) would be cleared",
        "Large file queue: {$largeCount} item(s) would be cleared",
        'Deduplication table: all entries would be cleared',
        'Progress records: all entries would be cleared',
        'Retry records: all entries would be cleared',
      ]);
      $this->io()->note('Dry run - no queues were cleared.');
      return;
    }

    $normalQueue->deleteQueue();
    $largeQueue->deleteQueue();

    // Clear deduplication table.
    $dedupeCount = $this->database->delete('ffmpeg_media_queued')->execute();

    // Clear orphaned progress records so items don't appear permanently stuck.
    $progressCount = $this->database->delete('ffmpeg_media_progress')->execute();

    // Clear retry records so re-queued items start with a fresh retry count.
    $retryCount = $this->database->delete('ffmpeg_media_retries')->execute();

    $this->logger()?->success('Cleared {normal} items from normal queue, {large} from large queue, {dedupe} dedupe entries, {progress} progress records, {retries} retry records.', [
      'normal' => $normalCount,
      'large' => $largeCount,
      'dedupe' => $dedupeCount,
      'progress' => $progressCount,
      'retries' => $retryCount,
    ]);
  }

  /**
   * Pause all queue processing globally.
   */
  #[CLI\Command(name: 'ffmpeg:queue-pause', aliases: ['ffqp'])]
  #[CLI\Usage(name: 'ffmpeg:queue-pause', description: 'Pause all video conversion queue processing.')]
  public function queuePause(): void {
    if ($this->queueScheduling->isGloballyPaused()) {
      $this->logger()?->warning('Queue processing is already paused.');
      return;
    }

    $this->queueScheduling->pauseQueue();
    $this->logger()?->success('Queue processing has been paused. No new conversions will start.');
  }

  /**
   * Resume queue processing after a global pause.
   */
  #[CLI\Command(name: 'ffmpeg:queue-resume', aliases: ['ffqr'])]
  #[CLI\Usage(name: 'ffmpeg:queue-resume', description: 'Resume video conversion queue processing.')]
  public function queueResume(): void {
    if (!$this->queueScheduling->isGloballyPaused()) {
      $this->logger()?->warning('Queue processing is not currently paused.');
      return;
    }

    $this->queueScheduling->resumeQueue();
    $this->logger()?->success('Queue processing has been resumed.');
  }

  /**
   * Processes queue items sequentially.
   *
   * @param int $limit
   *   Number of items to process.
   * @param string $queueName
   *   Queue name.
   */
  protected function processSequential(int $limit, string $queueName): void {
    $queue = $this->queueFactory->get($queueName);
    $config = $this->configFactory->get('ffmpeg_media.settings');
    $claimTimeout = (int) ($config->get('queue_claim_timeout') ?? 600);
    $processed = 0;
    $failed = 0;
    $skipped = 0;
    $stoppedEarly = FALSE;

    // Respect global pause — same check as the QueueWorker.
    if ($this->queueScheduling->isGloballyPaused()) {
      $this->logger()?->warning('Queue processing is globally paused. Use ffmpeg:queue-resume first.');
      return;
    }

    $progressBar = $this->io()->createProgressBar($limit);
    $progressBar->setFormat(' %current%/%max% [%bar%] %percent:3s%% %elapsed:6s%/%estimated:-6s% %message%');
    $progressBar->setMessage('Initializing...');
    $progressBar->start();

    while ($processed + $failed + $skipped < $limit) {
      $progressBar->setMessage('Waiting for queue item...');
      $progressBar->display();
      $item = $queue->claimItem($claimTimeout);

      if (!$item) {
        $stoppedEarly = TRUE;
        break;
      }

      $itemResult = $this->processClaimedItem($item->data, $queueName);
      if ($itemResult === 'requeue') {
        // Rotate to queue tail: create new item, then delete claimed one.
        $queue->createItem($item->data);
        $queue->deleteItem($item);
        $skipped++;
      }
      else {
        $queue->deleteItem($item);
        if ($itemResult === 'processed') {
          $processed++;
        }
        elseif ($itemResult === 'skipped') {
          $skipped++;
        }
        else {
          $failed++;
        }
      }
      $progressBar->advance();
    }

    $processedTotal = $processed + $failed + $skipped;

    if ($stoppedEarly && $processedTotal < $limit) {
      if ($processedTotal > 0) {
        $progressBar->setMaxSteps($processedTotal);
        $progressBar->setProgress($processedTotal);
      }
      $progressBar->setMessage('Stopped early - no more items.');
    }
    else {
      $progressBar->setMessage('Done!');
    }

    $progressBar->finish();
    $this->io()->newLine(2);

    $this->logger()?->success('Processed {processed} videos successfully. {failed} failed. {skipped} skipped.', [
      'processed' => $processed,
      'failed' => $failed,
      'skipped' => $skipped,
    ]);
  }

  /**
   * Processes a single claimed queue item with safety checks.
   *
   * Mirrors the safety semantics of VideoConversionQueueWorker::processItem():
   * lock acquisition, retry backoff, differentiated exception handling,
   * dead-letter storage, and progress tracking on failure.
   *
   * @param mixed $data
   *   Queue item data array.
   * @param string $queueName
   *   Queue name for per-item scheduling checks.
   *
   * @return string
   *   Result: 'processed', 'failed', 'skipped', or 'requeue'.
   *   'requeue' signals the caller to rotate the item to the queue tail
   *   (matching QueueWorker tail-rotation for backoff/lock/pause).
   */
  protected function processClaimedItem(mixed $data, string $queueName = ''): string {
    if (!is_array($data)) {
      return 'skipped';
    }

    $mediaId = (int) ($data['media_id'] ?? 0);
    $operation = $data['operation'] ?? 'convert';
    $profileId = $data['profile_id'] ?? NULL;
    $fieldName = (string) ($data['field_name'] ?? '');
    $selectedOutputs = $data['selected_outputs'] ?? [];
    $streamingOptions = $data['streaming_options'] ?? [];
    $itemKey = QueueItemKey::fromParts($mediaId, $fieldName);

    // Skip stale queue items removed from tracking via UI
    // (matches Worker line 165 — must run BEFORE isProcessable which
    // also returns false for untracked items, causing false requeue).
    if ($this->queueManager !== NULL && !$this->queueManager->isQueued($mediaId, $fieldName)) {
      return 'skipped';
    }

    // Per-item scheduling: global pause + Queue Editor pause/defer/priority
    // (matches Worker lines 177-186).
    if (!$this->queueScheduling->isProcessable($itemKey, $queueName)) {
      return 'requeue';
    }

    // Enforce retry backoff — rotate to tail so backoff can elapse
    // (matches Worker line 190 — before media load).
    if ($this->retryManager !== NULL
      && $this->retryManager->getRetryCount($mediaId, $fieldName) > 0
      && !$this->retryManager->isRetryDue($mediaId, $fieldName)) {
      return 'requeue';
    }

    // Load media entity (matches Worker line 204).
    $media = $this->entityTypeManager->getStorage('media')->load($mediaId);

    if (!$media instanceof MediaInterface) {
      $this->queueManager?->removeFromQueue($mediaId, $fieldName);
      return 'skipped';
    }

    // Acquire lock — rotate to tail if locked or concurrent limit reached.
    try {
      $lockAcquired = $this->lockManager?->acquireLock($mediaId) ?? TRUE;
      if (!$lockAcquired) {
        return 'requeue';
      }
    }
    catch (ConcurrentLimitException) {
      return 'requeue';
    }

    try {
      if ($operation === 'reconvert') {
        $this->videoConverter->deleteConvertedFiles($media, $fieldName);
      }

      $options = [
        'selected_outputs' => $selectedOutputs,
        'streaming_options' => $streamingOptions,
        'field_name' => $fieldName,
      ];

      $result = $this->videoConverter->convert($media, $profileId, FALSE, $options);

      if ($result->isSuccess()) {
        $this->retryManager?->resetRetries($mediaId, $fieldName);
        $this->queueManager?->removeFromQueue($mediaId, $fieldName);
        return 'processed';
      }

      // Soft failure from converter.
      $this->progressTracker?->markFailed($mediaId, $result->getError() ?? 'Conversion failed', $fieldName);
      $this->queueManager?->removeFromQueue($mediaId, $fieldName);
      return 'failed';
    }
    catch (ConcurrentLimitException) {
      // Concurrent limit reached during convert — requeue for later.
      return 'requeue';
    }
    catch (FFmpegException $e) {
      $this->storeDeadLetter($mediaId, $e->getMessage(), 0, $profileId, 'ffmpeg_error', $fieldName);
      $this->progressTracker?->markFailed($mediaId, $e->getMessage(), $fieldName);
      $this->queueManager?->removeFromQueue($mediaId, $fieldName);
      return 'failed';
    }
    catch (DatabaseException | \RuntimeException $e) {
      $this->retryManager?->recordRetry($mediaId, $e->getMessage(), $fieldName);
      if ($this->retryManager?->shouldRetry($mediaId, $fieldName) ?? FALSE) {
        return 'requeue';
      }
      $retryCount = $this->retryManager?->getRetryCount($mediaId, $fieldName) ?? 0;
      $this->storeDeadLetter($mediaId, $e->getMessage(), $retryCount, $profileId, 'max_retries_exceeded', $fieldName);
      $this->progressTracker?->markFailed($mediaId, $e->getMessage(), $fieldName);
      $this->queueManager?->removeFromQueue($mediaId, $fieldName);
      return 'failed';
    }
    catch (\Throwable $e) {
      $this->storeDeadLetter($mediaId, $e->getMessage(), 0, $profileId, 'unexpected_error', $fieldName);
      $this->progressTracker?->markFailed($mediaId, $e->getMessage(), $fieldName);
      $this->queueManager?->removeFromQueue($mediaId, $fieldName);
      return 'failed';
    }
    finally {
      $this->lockManager?->releaseLock($mediaId);
    }
  }

  /**
   * Stores a permanently failed job in dead-letter storage.
   *
   * @param int $mediaId
   *   The media entity ID.
   * @param string $errorMessage
   *   The error message.
   * @param int $retryCount
   *   Number of retry attempts made.
   * @param string|null $profileId
   *   Optional conversion profile ID.
   * @param string|null $lastErrorCode
   *   Optional error code for categorisation.
   * @param string $fieldName
   *   Source field name.
   */
  private function storeDeadLetter(
    int $mediaId,
    string $errorMessage,
    int $retryCount,
    ?string $profileId,
    ?string $lastErrorCode,
    string $fieldName = '',
  ): void {
    if ($this->deadLetterStorage === NULL) {
      return;
    }

    try {
      $this->deadLetterStorage->store($mediaId, $errorMessage, $retryCount, $profileId, $lastErrorCode, $fieldName);
    }
    catch (\Throwable $e) {
      $this->logger()?->error('Failed to store dead letter for media {id}: {error}', [
        'id' => $mediaId,
        'error' => $e->getMessage(),
      ]);
    }
  }

}
