<?php

declare(strict_types=1);

namespace Drupal\ffmpeg_media\Plugin\QueueWorker;

use Drupal\Core\Config\ConfigFactoryInterface;
use Drupal\Core\Database\DatabaseException;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Logger\LoggerChannelInterface;
use Drupal\Core\Plugin\ContainerFactoryPluginInterface;
use Drupal\Core\Queue\Attribute\QueueWorker;
use Drupal\Core\Queue\QueueFactory;
use Drupal\Core\Queue\QueueWorkerBase;
use Drupal\Core\Queue\RequeueException;
use Drupal\Core\Queue\SuspendQueueException;
use Drupal\Core\StringTranslation\TranslatableMarkup;
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 Symfony\Component\DependencyInjection\Attribute\Autowire;
use Symfony\Component\DependencyInjection\ContainerInterface;

/**
 * Queue worker for standard video conversion.
 */
#[QueueWorker(
  id: 'ffmpeg_media_conversion',
  title: new TranslatableMarkup('FFmpeg Media Conversion'),
  cron: ['time' => 60],
)]
class VideoConversionQueueWorker extends QueueWorkerBase implements ContainerFactoryPluginInterface {

  /**
   * Queue name (queue worker plugin ID).
   */
  protected readonly string $queueName;

  /**
   * Constructs a VideoConversionQueueWorker object.
   *
   * @param array $configuration
   *   Plugin configuration.
   * @param string $plugin_id
   *   Plugin ID.
   * @param mixed $plugin_definition
   *   Plugin definition.
   * @param \Drupal\ffmpeg_media\Service\VideoConverterInterface $videoConverter
   *   Video converter service.
   * @param \Drupal\ffmpeg_media\Service\QueueManagerInterface $queueManager
   *   Queue manager service.
   * @param \Drupal\ffmpeg_media\Service\LockManagerInterface $lockManager
   *   Lock manager service.
   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entityTypeManager
   *   Entity type manager.
   * @param \Drupal\Core\Logger\LoggerChannelInterface $logger
   *   Logger channel.
   * @param \Drupal\ffmpeg_media\Service\ProgressTrackerInterface $progressTracker
   *   Progress tracker service.
   * @param \Drupal\ffmpeg_media\Service\RetryManagerInterface $retryManager
   *   Retry manager service.
   * @param \Drupal\Core\Queue\QueueFactory $queueFactory
   *   Queue factory.
   * @param \Drupal\ffmpeg_media\Service\QueueSchedulingServiceInterface $queueScheduling
   *   Queue scheduling service.
   * @param \Drupal\ffmpeg_media\Service\DeadLetterStorageInterface|null $deadLetterStorage
   *   Dead-letter storage for permanently failed jobs.
   * @param \Drupal\Core\Config\ConfigFactoryInterface|null $configFactory
   *   Config factory for queue logging settings.
   */
  public function __construct(
    array $configuration,
    $plugin_id,
    $plugin_definition,
    #[Autowire(service: VideoConverterInterface::class)]
    protected readonly VideoConverterInterface $videoConverter,
    #[Autowire(service: QueueManagerInterface::class)]
    protected readonly QueueManagerInterface $queueManager,
    #[Autowire(service: LockManagerInterface::class)]
    protected readonly LockManagerInterface $lockManager,
    #[Autowire(service: 'entity_type.manager')]
    protected readonly EntityTypeManagerInterface $entityTypeManager,
    #[Autowire(service: 'logger.channel.ffmpeg_media')]
    protected readonly LoggerChannelInterface $logger,
    #[Autowire(service: ProgressTrackerInterface::class)]
    protected readonly ProgressTrackerInterface $progressTracker,
    #[Autowire(service: RetryManagerInterface::class)]
    protected readonly RetryManagerInterface $retryManager,
    #[Autowire(service: 'queue')]
    protected readonly QueueFactory $queueFactory,
    #[Autowire(service: QueueSchedulingServiceInterface::class)]
    protected readonly QueueSchedulingServiceInterface $queueScheduling,
    #[Autowire(service: DeadLetterStorageInterface::class)]
    protected readonly ?DeadLetterStorageInterface $deadLetterStorage = NULL,
    #[Autowire(service: 'config.factory')]
    protected readonly ?ConfigFactoryInterface $configFactory = NULL,
  ) {
    parent::__construct($configuration, $plugin_id, $plugin_definition);
    $this->queueName = (string) $plugin_id;
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  public static function create(
    ContainerInterface $container,
    array $configuration,
    $plugin_id,
    $plugin_definition,
  ): static {
    /** @phpstan-ignore-next-line Instantiated by Drupal plugin manager. */
    return new static(
      $configuration,
      $plugin_id,
      $plugin_definition,
      $container->get(VideoConverterInterface::class),
      $container->get(QueueManagerInterface::class),
      $container->get(LockManagerInterface::class),
      $container->get('entity_type.manager'),
      $container->get('logger.channel.ffmpeg_media'),
      $container->get(ProgressTrackerInterface::class),
      $container->get(RetryManagerInterface::class),
      $container->get('queue'),
      $container->get(QueueSchedulingServiceInterface::class),
      $container->get(DeadLetterStorageInterface::class),
      $container->get('config.factory'),
    );
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  public function processItem($data): void {
    $verboseQueueLogging = $this->isVerboseQueueLoggingEnabled();

    // Check concurrent limit.
    if ($this->queueManager->isLimitReached()) {
      throw new SuspendQueueException('Concurrent conversion limit reached.');
    }

    if (!is_array($data)) {
      $this->logger->error('Queue item data is not an array.');
      return;
    }

    $mediaId = $data['media_id'] ?? NULL;
    if ($mediaId === NULL) {
      $this->logger->error('Queue item missing media_id.');
      return;
    }
    $mediaId = (int) $mediaId;
    $fieldName = (string) ($data['field_name'] ?? '');
    $itemKey = QueueItemKey::fromParts($mediaId, $fieldName);

    // Skip stale queue items that were manually removed from tracking.
    if (!$this->queueManager->isQueued($mediaId, $fieldName)) {
      if ($verboseQueueLogging) {
        $this->logger->debug('Skipping stale queue item for media @id (no queue marker).', [
          '@id' => $mediaId,
        ]);
      }
      return;
    }

    // Apply queue editor scheduling (pause/manual order/priority).
    // Non-runnable items are rotated to the queue tail.
    if (!$this->queueScheduling->isProcessable($itemKey, $this->queueName)) {
      $this->queueFactory->get($this->queueName)->createItem($data);
      $state = $this->queueScheduling->isPaused($itemKey) ? 'paused' : 'deferred';
      if ($verboseQueueLogging) {
        $this->logger->debug('Queue item for media @id is @state by scheduler; moved to queue tail.', [
          '@id' => $mediaId,
          '@state' => $state,
        ]);
      }
      return;
    }

    // Enforce exponential backoff: skip items whose retry delay hasn't elapsed.
    if ($this->retryManager->getRetryCount($mediaId, $fieldName) > 0 && !$this->retryManager->isRetryDue($mediaId, $fieldName)) {
      // Rotate to queue tail; will be processed when backoff elapses.
      $this->queueFactory->get($this->queueName)->createItem($data);
      if ($verboseQueueLogging) {
        $nextRetry = $this->retryManager->getNextRetryTime($mediaId, $fieldName);
        $this->logger->debug('Retry backoff not elapsed for media @id (due at @time), moved to queue tail.', [
          '@id' => $mediaId,
          '@time' => date('Y-m-d H:i:s', $nextRetry),
        ]);
      }
      return;
    }

    // Load media entity.
    // Publication status is intentionally not checked: conversions are queued
    // by EntityHooks on insert/update — the authoring user already saved the
    // entity. Unpublished media (drafts, content moderation workflows) must
    // still be converted so outputs are ready when the item is published.
    $media = $this->entityTypeManager->getStorage('media')->load($mediaId);
    if (!$media instanceof MediaInterface) {
      $this->logger->warning('Media entity @id not found, removing from queue.', [
        '@id' => $mediaId,
      ]);
      $this->queueManager->removeFromQueue($mediaId, $fieldName);
      return;
    }

    // Try to acquire lock for this media entity.
    try {
      if (!$this->lockManager->acquireLock($mediaId)) {
        if ($verboseQueueLogging) {
          $this->logger->debug('Media @id is already being processed, requeueing.', [
            '@id' => $mediaId,
          ]);
        }
        throw new RequeueException('Media is already being processed.');
      }
    }
    catch (ConcurrentLimitException $e) {
      // Concurrent limit reached during lock acquisition.
      throw new RequeueException($e->getMessage());
    }

    $operation = $data['operation'] ?? 'convert';
    $profileId = $data['profile_id'] ?? NULL;
    $selectedOutputs = $data['selected_outputs'] ?? [];
    $streamingOptions = $data['streaming_options'] ?? [];

    if ($verboseQueueLogging) {
      $this->logger->debug('Starting @operation for media @id.', [
        '@operation' => $operation,
        '@id' => $mediaId,
      ]);
    }

    try {
      if ($operation === 'reconvert') {
        // Delete existing converted files before reconverting.
        $this->videoConverter->deleteConvertedFiles($media, $fieldName);
      }

      // Build conversion options.
      $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);
        if ($result->isPartialSuccess()) {
          $failedOutputs = $result->getFailedOutputs();
          $this->logger->warning('Conversion partially succeeded for media @id: @formats failed (@errors).', [
            '@id' => $mediaId,
            '@formats' => implode(', ', array_keys($failedOutputs)),
            '@errors' => implode('; ', $failedOutputs),
          ]);
        }
        elseif ($verboseQueueLogging) {
          $this->logger->debug('Conversion successful for media @id: @outputs', [
            '@id' => $mediaId,
            '@outputs' => implode(', ', array_keys($result->outputs)),
          ]);
        }
      }
      else {
        $this->logger->error('Conversion failed for media @id: @error', [
          '@id' => $mediaId,
          '@error' => $result->getError(),
        ]);
        // Defensive cleanup: ensure progress and queue markers are cleared
        // even if convert() returned a soft failure without full cleanup.
        $this->progressTracker->markFailed($mediaId, $result->getError() ?? 'Conversion failed', $fieldName);
        $this->queueManager->removeFromQueue($mediaId, $fieldName);
      }
    }
    catch (ConcurrentLimitException $e) {
      // Requeue for later processing.
      throw new RequeueException($e->getMessage());
    }
    catch (FFmpegException $e) {
      // Permanent conversion failure (codec error, invalid input, etc.).
      $this->logger->error('FFmpeg conversion failed for media @id: @error', [
        '@id' => $mediaId,
        '@error' => $e->getMessage(),
      ]);
      $this->storeDeadLetter($mediaId, $e->getMessage(), 0, $profileId, 'ffmpeg_error', $fieldName);
      $this->progressTracker->markFailed($mediaId, $e->getMessage(), $fieldName);
      $this->queueManager->removeFromQueue($mediaId, $fieldName);
    }
    catch (DatabaseException | \RuntimeException $e) {
      // Transient failures (database connectivity, filesystem I/O).
      $this->retryManager->recordRetry($mediaId, $e->getMessage(), $fieldName);
      if ($this->retryManager->shouldRetry($mediaId, $fieldName)) {
        $this->logger->warning('Transient error for media @id (will retry): @error', [
          '@id' => $mediaId,
          '@error' => $e->getMessage(),
        ]);
        // Rotate to queue tail instead of throwing RequeueException, so the
        // backoff check at the top of processItem() can enforce the delay.
        $this->queueFactory->get($this->queueName)->createItem($data);
        if ($verboseQueueLogging) {
          $this->logger->debug('Transient error item for media @id moved to queue tail for backoff.', [
            '@id' => $mediaId,
          ]);
        }
        return;
      }
      // Max retries exceeded - treat as permanent failure.
      $this->logger->error('Max retries exceeded for media @id: @error', [
        '@id' => $mediaId,
        '@error' => $e->getMessage(),
      ]);
      $retryCount = $this->retryManager->getRetryCount($mediaId, $fieldName);
      $this->storeDeadLetter($mediaId, $e->getMessage(), $retryCount, $profileId, 'max_retries_exceeded', $fieldName);
      $this->progressTracker->markFailed($mediaId, $e->getMessage(), $fieldName);
      $this->queueManager->removeFromQueue($mediaId, $fieldName);
    }
    catch (\Throwable $e) {
      // Truly unexpected error - log critical, mark failed, remove.
      $this->logger->critical('Unexpected error processing media @id: @error', [
        '@id' => $mediaId,
        '@error' => $e->getMessage(),
      ]);
      $this->storeDeadLetter($mediaId, $e->getMessage(), 0, $profileId, 'unexpected_error', $fieldName);
      $this->progressTracker->markFailed($mediaId, $e->getMessage(), $fieldName);
      $this->queueManager->removeFromQueue($mediaId, $fieldName);
    }
    finally {
      // Always release the lock when done.
      $this->lockManager->releaseLock($mediaId);
    }
  }

  /**
   * Stores a permanently failed job in dead-letter storage.
   *
   * Catches exceptions to avoid masking the original conversion failure,
   * but logs at critical level so dead-letter storage problems surface
   * prominently in monitoring.
   *
   * @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 for field-scoped dead-letter storage.
   */
  private function storeDeadLetter(int $mediaId, string $errorMessage, int $retryCount, ?string $profileId, ?string $lastErrorCode, string $fieldName = ''): void {
    if ($this->deadLetterStorage === NULL) {
      $this->logger->warning('Dead-letter storage unavailable; permanently failed conversion for media @id will not be recorded.', [
        '@id' => $mediaId,
      ]);
      return;
    }

    try {
      $this->deadLetterStorage->store($mediaId, $errorMessage, $retryCount, $profileId, $lastErrorCode, $fieldName);
    }
    catch (\Throwable $deadLetterException) {
      $this->logger->critical('Failed to store dead letter for media @id: @error. The original conversion error was: @original', [
        '@id' => $mediaId,
        '@error' => $deadLetterException->getMessage(),
        '@original' => $errorMessage,
      ]);
    }
  }

  /**
   * Returns whether verbose queue lifecycle logging is enabled.
   */
  private function isVerboseQueueLoggingEnabled(): bool {
    // Default when tests instantiate without config factory.
    if ($this->configFactory === NULL) {
      return TRUE;
    }

    return (bool) ($this->configFactory
      ->get('ffmpeg_media.settings')
      ->get('queue_verbose_logging') ?? FALSE);
  }

}
