<?php

declare(strict_types=1);

namespace Drupal\ffmpeg_media\Service;

use Drupal\Core\Cache\CacheTagsInvalidatorInterface;
use Drupal\Core\Config\ConfigFactoryInterface;
use Drupal\Core\Database\Connection;
use Drupal\Core\Database\IntegrityConstraintViolationException;
use Drupal\Core\Logger\LoggerChannelInterface;
use Drupal\Core\Queue\QueueFactory;
use Drupal\ffmpeg_media\Enum\ConversionStatus;
use Drupal\ffmpeg_media\Exception\ConcurrentLimitException;
use Drupal\ffmpeg_media\Utility\QueueItemKey;
use Drupal\file\FileInterface;
use Drupal\media\MediaInterface;

/**
 * Service for managing the video conversion queue.
 *
 * @internal
 */
final class QueueManager implements QueueManagerInterface {

  /**
   * Table name for queue deduplication.
   */
  protected const string TABLE = 'ffmpeg_media_queued';

  /**
   * Video converter factory closure.
   *
   * @var (\Closure(): VideoConverterInterface)|null
   */
  protected ?\Closure $videoConverterFactory = NULL;

  /**
   * Resolved video converter instance.
   */
  protected ?VideoConverterInterface $videoConverter = NULL;

  /**
   * Constructs a QueueManager object.
   *
   * @param \Drupal\Core\Queue\QueueFactory $queueFactory
   *   Queue factory.
   * @param \Drupal\Core\Database\Connection $database
   *   Database connection.
   * @param \Drupal\Core\Config\ConfigFactoryInterface $configFactory
   *   Config factory.
   * @param \Drupal\ffmpeg_media\Service\ProgressTrackerInterface $progressTracker
   *   Progress tracker service.
   * @param \Drupal\ffmpeg_media\Service\LockManagerInterface $lockManager
   *   Lock manager service.
   * @param \Drupal\Core\Logger\LoggerChannelInterface $logger
   *   Logger channel.
   * @param \Drupal\ffmpeg_media\Service\QueueAssignmentRepositoryInterface $queueAssignmentRepository
   *   Queue assignment metadata repository.
   * @param \Drupal\ffmpeg_media\Service\QueueAssignmentServiceInterface $queueAssignmentService
   *   Queue assignment service for routing, context resolution, and dedup.
   * @param \Drupal\Core\Cache\CacheTagsInvalidatorInterface|null $cacheTagsInvalidator
   *   Optional cache tags invalidator for live dashboard views.
   */
  public function __construct(
    protected readonly QueueFactory $queueFactory,
    protected readonly Connection $database,
    protected readonly ConfigFactoryInterface $configFactory,
    protected readonly ProgressTrackerInterface $progressTracker,
    protected readonly LockManagerInterface $lockManager,
    protected readonly LoggerChannelInterface $logger,
    protected readonly QueueAssignmentRepositoryInterface $queueAssignmentRepository,
    protected readonly QueueAssignmentServiceInterface $queueAssignmentService,
    protected readonly ?CacheTagsInvalidatorInterface $cacheTagsInvalidator = NULL,
  ) {}

  /**
   * Sets the video converter service via closure for lazy resolution.
   *
   * Uses a closure to break the circular dependency between
   * QueueManager and VideoConverter at container compilation time.
   *
   * @phpstan-param \Closure(): VideoConverterInterface $videoConverterFactory
   */
  public function setVideoConverter(\Closure $videoConverterFactory): void {
    $this->videoConverterFactory = $videoConverterFactory;
  }

  /**
   * Gets the video converter service.
   *
   * Lazily resolves the service on first access.
   *
   * @return \Drupal\ffmpeg_media\Service\VideoConverterInterface
   *   The video converter service.
   *
   * @throws \RuntimeException
   *   If video converter factory is not set.
   */
  protected function getVideoConverter(): VideoConverterInterface {
    if ($this->videoConverter === NULL) {
      if ($this->videoConverterFactory === NULL) {
        throw new \RuntimeException('VideoConverter service has not been set.');
      }
      $this->videoConverter = ($this->videoConverterFactory)();
    }
    if (!$this->videoConverter instanceof VideoConverterInterface) {
      throw new \RuntimeException('VideoConverter service closure did not resolve to a VideoConverterInterface instance.');
    }
    return $this->videoConverter;
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  public function queueForConversion(
    MediaInterface $media,
    string $operation = 'convert',
    ?string $profileId = NULL,
    array $selectedOutputs = [],
    array $streamingOptions = [],
    string $fieldName = '',
  ): bool {
    $context = $this->queueAssignmentService->resolveConversionContext($media, $fieldName);
    if ($context === NULL) {
      return FALSE;
    }

    $mediaId = $context['media_id'];
    $fieldName = $context['field_name'];
    $sourceFile = $context['source_file'];

    $queueTarget = $this->queueAssignmentService->resolveQueueTarget($sourceFile, $mediaId, $fieldName);
    if ($queueTarget === NULL) {
      return FALSE;
    }

    return $this->insertQueueEntry(
      mediaId: $mediaId,
      fieldName: $fieldName,
      queueName: $queueTarget['queue_name'],
      operation: $operation,
      profileId: $profileId,
      selectedOutputs: $selectedOutputs,
      streamingOptions: $streamingOptions,
    );
  }

  /**
   * Inserts the queue entry with dedup handling and rollback on failure.
   *
   * Atomically inserts a dedup marker, persists the queue assignment,
   * starts progress tracking, and creates the queue item. On failure,
   * compensating actions clean up partial state.
   *
   * @param int $mediaId
   *   The media entity ID.
   * @param string $fieldName
   *   The field name.
   * @param string $queueName
   *   The target queue name.
   * @param string $operation
   *   The operation type.
   * @param string|null $profileId
   *   Optional conversion profile ID.
   * @param array<int, string> $selectedOutputs
   *   Selected output formats.
   * @param array<string, mixed> $streamingOptions
   *   Streaming options.
   *
   * @return bool
   *   TRUE if the item was successfully queued, FALSE otherwise.
   */
  private function insertQueueEntry(
    int $mediaId,
    string $fieldName,
    string $queueName,
    string $operation,
    ?string $profileId,
    array $selectedOutputs,
    array $streamingOptions,
  ): bool {
    // Atomically insert into the dedup table. The composite primary key
    // constraint prevents duplicates without a TOCTOU race condition.
    // If another process already inserted this queue item, the constraint
    // violation tells us it was already queued.
    try {
      $this->database->insert(self::TABLE)
        ->fields([
          'media_id' => $mediaId,
          'field_name' => $fieldName,
          'queued_at' => time(),
          'profile_id' => $profileId,
        ])
        ->execute();
    }
    catch (IntegrityConstraintViolationException) {
      // Another process already queued this media ID.
      return FALSE;
    }

    $itemId = QueueItemKey::fromParts($mediaId, $fieldName);

    // Create queue item with extended options.
    $queue = $this->queueFactory->get($queueName);
    $item = [
      'media_id' => $mediaId,
      'operation' => $operation,
      'profile_id' => $profileId ?? $this->getConfig('default_profile'),
      'selected_outputs' => $selectedOutputs,
      'streaming_options' => $streamingOptions,
      'field_name' => $fieldName,
      'queued_at' => time(),
    ];

    try {
      $this->queueAssignmentRepository->persistAssignment($itemId, $queueName);
      $this->progressTracker->startTracking($mediaId, NULL, $fieldName);
      $queue->createItem($item);
    }
    catch (\Throwable $e) {
      // Clean up progress tracking on failure.
      try {
        $this->progressTracker->deleteProgress($mediaId, $fieldName);
      }
      catch (\Throwable) {
        // Ignore cleanup failures.
      }

      // Compensate dedup insert when queue item creation fails.
      try {
        $this->removeFromQueue($mediaId, $fieldName);
      }
      catch (\Throwable $rollbackException) {
        $this->logger->error('Failed to rollback queue marker for media @id after queue insert failure: @error', [
          '@id' => $mediaId,
          '@error' => $rollbackException->getMessage(),
        ]);
      }

      $this->logger->error('Failed to create queue item for media @id: @error', [
        '@id' => $mediaId,
        '@error' => $e->getMessage(),
      ]);
      return FALSE;
    }

    $this->invalidateLiveCaches();

    return TRUE;
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  #[\NoDiscard]
  public function isQueued(int $mediaId, string $fieldName = ''): bool {
    $query = $this->database->select(self::TABLE, 'q')
      ->condition('media_id', $mediaId);

    if ($fieldName !== '') {
      $query->condition('field_name', [$fieldName, ''], 'IN');
    }

    $statement = $query->countQuery()->execute();

    return (bool) ($statement?->fetchField() ?? FALSE);
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  #[\NoDiscard]
  public function getQueuedMediaCount(): int {
    $statement = $this->database->select(self::TABLE, 'q')
      ->countQuery()
      ->execute();

    return (int) ($statement?->fetchField() ?? 0);
  }

  /**
   * {@inheritdoc}
   *
   * Note: this intentionally only removes the dedup marker from
   * ffmpeg_media_queued, not the Drupal queue item itself. The
   * QueueWorker validates isQueued() before processing, so orphaned
   * queue items are harmlessly skipped when claimed. This avoids the
   * complexity and race conditions of scanning the queue table for
   * matching items.
   */
  #[\Override]
  public function removeFromQueue(int $mediaId, string $fieldName = ''): void {
    if ($fieldName === '') {
      $this->queueAssignmentRepository->clearAssignmentsForMedia($mediaId);
    }
    else {
      $this->queueAssignmentRepository->clearAssignment(QueueItemKey::fromParts($mediaId, $fieldName));
      $this->queueAssignmentRepository->clearAssignment(QueueItemKey::fromParts($mediaId));
    }

    $delete = $this->database->delete(self::TABLE)
      ->condition('media_id', $mediaId);

    if ($fieldName !== '') {
      $delete->condition('field_name', [$fieldName, ''], 'IN');
    }

    $delete->execute();
    $this->invalidateLiveCaches();
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  #[\NoDiscard]
  public function getQueuedMediaIds(): array {
    $statement = $this->database->select(self::TABLE, 'q')
      ->fields('q', ['media_id', 'field_name'])
      ->orderBy('queued_at', 'ASC')
      ->orderBy('media_id', 'ASC')
      ->orderBy('field_name', 'ASC')
      ->execute();

    if ($statement === NULL) {
      return [];
    }

    $ids = [];
    foreach ($statement as $row) {
      $mediaId = (int) ($row->media_id ?? 0);
      if ($mediaId > 0) {
        $ids[$mediaId] = $mediaId;
      }
    }

    return array_values($ids);
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  #[\NoDiscard]
  public function getQueuedItemIds(): array {
    $statement = $this->database->select(self::TABLE, 'q')
      ->fields('q', ['media_id', 'field_name'])
      ->orderBy('queued_at', 'ASC')
      ->orderBy('media_id', 'ASC')
      ->orderBy('field_name', 'ASC')
      ->execute();

    if ($statement === NULL) {
      return [];
    }

    $itemIds = [];
    foreach ($statement as $row) {
      $mediaId = (int) ($row->media_id ?? 0);
      if ($mediaId <= 0) {
        continue;
      }

      $itemIds[] = QueueItemKey::fromParts($mediaId, (string) ($row->field_name ?? ''));
    }

    return $itemIds;
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  #[\NoDiscard]
  public function getQueuedMediaIdsForQueue(string $queueName): array {
    $itemIds = $this->getQueuedItemIdsForQueue($queueName);
    if ($itemIds === []) {
      return [];
    }

    $mediaIds = [];
    foreach ($itemIds as $itemId) {
      $mediaId = QueueItemKey::parse($itemId)['media_id'];
      if ($mediaId > 0) {
        $mediaIds[$mediaId] = $mediaId;
      }
    }

    return array_values($mediaIds);
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  #[\NoDiscard]
  public function getQueuedItemIdsForQueue(string $queueName): array {
    if ($queueName === '') {
      return [];
    }

    $queuedItemIds = $this->getQueuedItemIds();
    if ($queuedItemIds === []) {
      return [];
    }

    $assignments = $this->queueAssignmentRepository->getAssignments($queuedItemIds);
    $resolved = [];

    foreach ($queuedItemIds as $itemId) {
      $assignedQueue = $assignments[$itemId] ?? NULL;
      if (!is_string($assignedQueue) || $assignedQueue === '' || $assignedQueue === $queueName) {
        $resolved[] = $itemId;
      }
    }

    return $resolved;
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  #[\NoDiscard]
  public function getQueueStats(): array {
    if (!$this->queueAssignmentRepository->isAvailable()) {
      $standardQueue = $this->queueFactory->get(self::QUEUE_NAME);
      $largeQueue = $this->queueFactory->get(self::QUEUE_LARGE_NAME);
      $totalQueued = $standardQueue->numberOfItems() + $largeQueue->numberOfItems();
      $processing = $this->progressTracker->getActiveCount();

      $statement = $this->database->select(self::TABLE, 'q')
        ->fields('q', ['queued_at'])
        ->orderBy('queued_at', 'ASC')
        ->range(0, 1)
        ->execute();
      $oldest = $statement?->fetchField() ?? FALSE;

      return [
        'total_queued' => $totalQueued,
        'pending' => max(0, $totalQueued - $processing),
        'processing' => $processing,
        'oldest_item' => $oldest !== FALSE ? (int) $oldest : NULL,
        'standard_queue' => $standardQueue->numberOfItems(),
        'large_queue' => $largeQueue->numberOfItems(),
      ];
    }

    $queuedItemIds = $this->getQueuedItemIds();
    $standardQueued = $this->getQueuedItemIdsForQueue(self::QUEUE_NAME);
    $largeQueued = $this->getQueuedItemIdsForQueue(self::QUEUE_LARGE_NAME);
    $totalQueued = count($queuedItemIds);
    $processing = $this->progressTracker->getActiveCount();

    $statement = $this->database->select(self::TABLE, 'q')
      ->fields('q', ['queued_at'])
      ->orderBy('queued_at', 'ASC')
      ->range(0, 1)
      ->execute();
    $oldest = $statement?->fetchField() ?? FALSE;

    return [
      'total_queued' => $totalQueued,
      'pending' => max(0, $totalQueued - $processing),
      'processing' => $processing,
      'oldest_item' => $oldest !== FALSE ? (int) $oldest : NULL,
      'standard_queue' => count($standardQueued),
      'large_queue' => count($largeQueued),
    ];
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  #[\NoDiscard]
  public function getQueueNameForSize(int $fileSize): string {
    return $this->queueAssignmentService->getQueueNameForSize($fileSize);
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  #[\NoDiscard]
  public function isLimitReached(): bool {
    $maxConcurrent = $this->getConfig('max_concurrent') ?? 2;
    $activeCount = $this->progressTracker->getActiveCount();

    return $activeCount >= $maxConcurrent;
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  #[\NoDiscard]
  public function getQueueForFile(FileInterface $file): string {
    return $this->queueAssignmentService->getQueueForFile($file);
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  #[\NoDiscard]
  public function getLargeFileThreshold(): int {
    return $this->queueAssignmentService->getLargeFileThreshold();
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  public function cleanupStuckConversions(): int {
    $timeout = (int) ($this->getConfig('stuck_process_timeout') ?? 7200);
    $staleThreshold = time() - $timeout;

    $stuckConversions = $this->progressTracker->getStuckConversions($staleThreshold);

    $cleanedUp = 0;
    foreach ($stuckConversions as $progress) {
      // Release any locks.
      $this->lockManager->releaseLock($progress->mediaId);

      // Mark as failed.
      $this->progressTracker->markFailed(
        $progress->mediaId,
        'Conversion timed out after ' . $timeout . ' seconds',
        $progress->fieldName,
      );

      // Remove from queue tracking.
      $this->removeFromQueue($progress->mediaId, $progress->fieldName);

      $this->logger->warning('Cleaned up stuck conversion for media @id', [
        '@id' => $progress->mediaId,
      ]);

      $cleanedUp++;
    }

    // Clean up expired locks separately (not counted in stuck cleanup).
    $this->lockManager->cleanupExpiredLocks();

    return $cleanedUp;
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  public function processImmediately(MediaInterface $media, ?string $profileId = NULL, string $fieldName = ''): bool {
    $mediaId = (int) $media->id();
    $conversionOptions = [];
    if ($fieldName !== '') {
      $conversionOptions['field_name'] = $fieldName;
    }

    // Check if conversion is already running.
    if ($this->lockManager->hasLock($mediaId)) {
      $this->logger->warning('Cannot process media @id immediately: conversion already in progress.', [
        '@id' => $mediaId,
      ]);
      return FALSE;
    }

    // Acquire lock.
    try {
      if (!$this->lockManager->acquireLock($mediaId)) {
        $this->logger->warning('Cannot process media @id immediately: failed to acquire lock.', [
          '@id' => $mediaId,
        ]);
        return FALSE;
      }
    }
    catch (ConcurrentLimitException $e) {
      $this->logger->warning('Cannot process media @id immediately: concurrent limit reached.', [
        '@id' => $mediaId,
      ]);
      return FALSE;
    }

    try {
      // Call VideoConverter directly (not via queue) using synchronous
      // execution and the current field context.
      $result = $this->getVideoConverter()->convert($media, $profileId, TRUE, $conversionOptions);

      if ($this->isVerboseQueueLoggingEnabled()) {
        $this->logger->debug('Immediate conversion completed for media @id with status: @status', [
          '@id' => $mediaId,
          '@status' => $result->isSuccess() ? 'success' : 'failed',
        ]);
      }

      return $result->isSuccess();
    }
    catch (\Throwable $e) {
      // Clean up the matching dedup marker so the media can be re-queued
      // after failure without disturbing other field-specific work.
      try {
        $this->removeFromQueue($mediaId, $fieldName);
      }
      catch (\Throwable) {
        // Ignore cleanup failures.
      }

      $this->logger->error('Immediate conversion failed for media @id: @message', [
        '@id' => $mediaId,
        '@message' => $e->getMessage(),
      ]);
      return FALSE;
    }
    finally {
      // Always release lock.
      $this->lockManager->releaseLock($mediaId);
    }
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  public function cleanupStaleQueueEntries(int $maxAge): int {
    $cutoff = time() - $maxAge;

    $statement = $this->database->select(self::TABLE, 'q')
      ->fields('q', ['media_id', 'field_name'])
      ->condition('queued_at', $cutoff, '<')
      ->execute();
    $staleRows = $statement?->fetchAll() ?? [];

    if ($staleRows === []) {
      return 0;
    }

    $staleByItem = [];
    $staleMediaIds = [];
    foreach ($staleRows as $row) {
      $mediaId = (int) ($row->media_id ?? 0);
      if ($mediaId <= 0) {
        continue;
      }

      $fieldName = (string) ($row->field_name ?? '');
      $itemId = QueueItemKey::fromParts($mediaId, $fieldName);
      $staleByItem[$itemId] = [
        'media_id' => $mediaId,
        'field_name' => $fieldName,
      ];
      $staleMediaIds[$mediaId] = $mediaId;
    }

    if ($staleByItem === []) {
      return 0;
    }

    // Exclude items still actively pending or converting. Their dedup
    // markers must remain so the queue worker does not skip them.
    $activeStatement = $this->database->select('ffmpeg_media_progress', 'p')
      ->fields('p', ['media_id', 'field_name'])
      ->condition('media_id', array_values($staleMediaIds), 'IN')
      ->condition('status', [
        ConversionStatus::Pending->value,
        ConversionStatus::Converting->value,
      ], 'IN')
      ->execute();
    $activeItemIds = [];
    if ($activeStatement !== NULL) {
      $activeRows = $activeStatement->fetchAll() ?: [];
      foreach ($activeRows as $row) {
        $activeItemIds[] = QueueItemKey::fromParts(
          (int) ($row->media_id ?? 0),
          (string) ($row->field_name ?? ''),
        );
      }
    }

    $toDelete = array_values(array_diff(array_keys($staleByItem), $activeItemIds));

    if ($toDelete === []) {
      return 0;
    }

    $deleted = 0;
    $legacyMediaIds = [];
    $scopedItemIds = [];
    foreach ($toDelete as $itemId) {
      $fieldName = $staleByItem[$itemId]['field_name'];
      if ($fieldName === '') {
        $legacyMediaIds[] = $staleByItem[$itemId]['media_id'];
        continue;
      }
      $scopedItemIds[] = $itemId;
    }

    if ($legacyMediaIds !== []) {
      $deleted += (int) $this->database->delete(self::TABLE)
        ->condition('media_id', array_values(array_unique($legacyMediaIds)), 'IN')
        ->condition('field_name', '')
        ->execute();
    }

    foreach ($scopedItemIds as $itemId) {
      $deleted += (int) $this->database->delete(self::TABLE)
        ->condition('media_id', $staleByItem[$itemId]['media_id'])
        ->condition('field_name', $staleByItem[$itemId]['field_name'])
        ->execute();
    }

    if ($deleted > 0) {
      $this->queueAssignmentRepository->clearAssignments($toDelete);
      $this->invalidateLiveCaches();
    }

    return $deleted;
  }

  /**
   * Gets a config value.
   *
   * @param string $key
   *   Config key.
   *
   * @return mixed
   *   Config value.
   */
  protected function getConfig(string $key): mixed {
    return $this->configFactory->get('ffmpeg_media.settings')->get($key);
  }

  /**
   * Returns whether verbose queue lifecycle logging is enabled.
   */
  private function isVerboseQueueLoggingEnabled(): bool {
    return (bool) ($this->getConfig('queue_verbose_logging') ?? FALSE);
  }

  /**
   * Invalidates dashboard and stats cache tags after queue changes.
   */
  private function invalidateLiveCaches(): void {
    $this->cacheTagsInvalidator?->invalidateTags([
      'ffmpeg_media:stats',
      'ffmpeg_media:dashboard',
    ]);
  }

}
