<?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\ffmpeg_media\Enum\ConversionStatus;
use Drupal\ffmpeg_media\ValueObject\ProgressData;

/**
 * Service for tracking video conversion progress.
 *
 * @internal
 */
final class ProgressTracker implements ProgressTrackerInterface {

  /**
   * Table name for progress tracking.
   */
  protected const string TABLE = 'ffmpeg_media_progress';

  /**
   * Cache TTL for active count in seconds.
   */
  private const float ACTIVE_COUNT_CACHE_TTL = 5.0;

  /**
   * Default minimum interval between DB writes for the same media_id.
   */
  private const int DEFAULT_PROGRESS_THROTTLE_INTERVAL = 5;

  /**
   * Status values that represent transitions and must always be written.
   *
   * @var array<int, \Drupal\ffmpeg_media\Enum\ConversionStatus>
   */
  private const array STATUS_TRANSITION_VALUES = [
    ConversionStatus::Pending,
    ConversionStatus::Completed,
    ConversionStatus::Failed,
  ];

  /**
   * In-memory throttle state cache to avoid per-update DB queries.
   *
   * Keyed by "{mediaId}:{fieldName}". Populated on first access per media,
   * then reused for subsequent throttle checks within the same request.
   *
   * @var array<string, array{status: string, last_write: int|null}|null>
   */
  private array $throttleStateCache = [];

  /**
   * Cached active conversion count.
   */
  private ?int $cachedActiveCount = NULL;

  /**
   * Timestamp when active count was last cached.
   */
  private float $activeCountCachedAt = 0.0;

  /**
   * Constructs a ProgressTracker object.
   *
   * @param \Drupal\Core\Database\Connection $database
   *   Database connection.
   * @param \Drupal\Core\Config\ConfigFactoryInterface $configFactory
   *   Config factory for throttle settings.
   * @param \Drupal\Core\Cache\CacheTagsInvalidatorInterface|null $cacheTagsInvalidator
   *   Optional cache tags invalidator for live dashboard views.
   */
  public function __construct(
    protected readonly Connection $database,
    protected readonly ConfigFactoryInterface $configFactory,
    protected readonly ?CacheTagsInvalidatorInterface $cacheTagsInvalidator = NULL,
  ) {}

  /**
   * {@inheritdoc}
   */
  #[\Override]
  public function startTracking(int $mediaId, ?string $correlationId = NULL, string $fieldName = ''): void {
    $now = time();

    $this->database->merge(self::TABLE)
      ->keys([
        'media_id' => $mediaId,
        'field_name' => $fieldName,
      ])
      ->fields([
        'field_name' => $fieldName,
        'status' => ConversionStatus::Pending->value,
        'progress_percent' => 0,
        'current_step' => NULL,
        'current_output' => NULL,
        'error_message' => NULL,
        'started' => $now,
        'updated' => $now,
        'completed' => NULL,
        'correlation_id' => $correlationId,
        'last_write' => NULL,
      ])
      ->execute();
    $this->invalidateActiveCountCache();
    $this->invalidateLiveCaches($mediaId);
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  public function updateProgress(
    int $mediaId,
    ConversionStatus $status,
    int $percent,
    ?string $step = NULL,
    ?string $currentOutput = NULL,
    ?float $elapsedSeconds = NULL,
    string $fieldName = '',
  ): void {
    // Status changes (e.g., pending->converting, converting->completed/failed)
    // must always be written immediately. Only throttle same-status progress
    // updates to reduce DB writes during high-concurrency conversions.
    if (!$this->shouldWriteProgress($mediaId, $status, $fieldName)) {
      return;
    }

    $targetFieldName = $this->getWriteTargetFieldName($mediaId, $fieldName);

    // Use transaction + SELECT FOR UPDATE to prevent lost updates when
    // two workers update the same media_id concurrently.
    $transaction = $this->database->startTransaction();
    try {
      $lockQuery = $this->database->select(self::TABLE, 'p')
        ->fields('p', ['media_id'])
        ->condition('media_id', $mediaId)
        ->condition('field_name', $targetFieldName);
      $lockQuery->forUpdate();
      $lockQuery->execute();

      $now = time();
      $fields = [
        'status' => $status->value,
        'progress_percent' => min(100, max(0, $percent)),
        'current_step' => $step,
        'current_output' => $currentOutput,
        'updated' => $now,
        'last_write' => $now,
      ];
      if ($elapsedSeconds !== NULL) {
        $fields['elapsed_seconds'] = $elapsedSeconds;
      }
      $this->database->update(self::TABLE)
        ->fields($fields)
        ->condition('media_id', $mediaId)
        ->condition('field_name', $targetFieldName)
        ->execute();
    }
    catch (\Throwable $e) {
      $transaction->rollBack();
      throw new \RuntimeException(
        sprintf('Failed to update progress for media %d (field: %s, status: %s): %s', $mediaId, $targetFieldName, $status->value, $e->getMessage()),
        0,
        $e,
      );
    }
    unset($transaction);

    // Invalidate throttle cache for this media so the next check reads fresh.
    $throttleCacheKey = $mediaId . ':' . $targetFieldName;
    unset($this->throttleStateCache[$throttleCacheKey]);

    $this->invalidateActiveCountCache();
    $this->invalidateLiveCaches($mediaId);
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  public function markCompleted(int $mediaId, string $fieldName = ''): void {
    $now = time();
    $targetFieldName = $this->getWriteTargetFieldName($mediaId, $fieldName);

    $this->database->update(self::TABLE)
      ->fields([
        'status' => ConversionStatus::Completed->value,
        'progress_percent' => 100,
        'current_step' => NULL,
        'current_output' => NULL,
        'error_message' => NULL,
        'updated' => $now,
        'completed' => $now,
      ])
      ->condition('media_id', $mediaId)
      ->condition('field_name', $targetFieldName)
      ->execute();
    $this->invalidateActiveCountCache();
    $this->invalidateLiveCaches($mediaId);
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  public function markFailed(int $mediaId, string $error, string $fieldName = ''): void {
    $now = time();
    $targetFieldName = $this->getWriteTargetFieldName($mediaId, $fieldName);

    $this->database->merge(self::TABLE)
      ->keys([
        'media_id' => $mediaId,
        'field_name' => $targetFieldName,
      ])
      ->insertFields([
        'media_id' => $mediaId,
        'field_name' => $targetFieldName,
        'status' => ConversionStatus::Failed->value,
        'progress_percent' => 0,
        'error_message' => $error,
        'started' => $now,
        'updated' => $now,
        'completed' => $now,
      ])
      ->updateFields([
        'status' => ConversionStatus::Failed->value,
        'error_message' => $error,
        'updated' => $now,
        'completed' => $now,
      ])
      ->execute();
    $this->invalidateActiveCountCache();
    $this->invalidateLiveCaches($mediaId);
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  #[\NoDiscard]
  public function getProgress(int $mediaId, string $fieldName = ''): ?ProgressData {
    $row = $this->getProgressRow($mediaId, $fieldName);
    if ($row === NULL) {
      return NULL;
    }

    return ProgressData::fromDatabaseRow($row);
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  #[\NoDiscard]
  public function getMultipleProgress(array $mediaIds): array {
    if ($mediaIds === []) {
      return [];
    }

    $uniqueIds = array_values(array_unique($mediaIds));

    $statement = $this->database->select(self::TABLE, 'p')
      ->fields('p')
      ->condition('media_id', $uniqueIds, 'IN')
      ->execute();

    if ($statement === NULL) {
      return array_fill_keys($uniqueIds, NULL);
    }

    // Group all rows by media_id.
    /** @var array<int, list<array<string, int|string|null>>> $grouped */
    $grouped = [];
    foreach ($statement as $row) {
      $row = (array) $row;
      $id = (int) ($row['media_id'] ?? 0);
      $grouped[$id][] = $row;
    }

    // For each media_id, pick the best-fit row using the same priority
    // logic as getProgressRow(): converting > pending > failed > completed,
    // then most recently updated.
    $priority = static fn(string $status): int => match ($status) {
      ConversionStatus::Converting->value => 0,
      ConversionStatus::Pending->value => 1,
      ConversionStatus::Failed->value => 2,
      ConversionStatus::Completed->value => 3,
      default => 4,
    };

    $result = [];
    foreach ($uniqueIds as $id) {
      if (!isset($grouped[$id]) || $grouped[$id] === []) {
        $result[$id] = NULL;
        continue;
      }

      $rows = $grouped[$id];
      usort($rows, static function (array $a, array $b) use ($priority): int {
        $statusComparison = $priority((string) ($a['status'] ?? '')) <=> $priority((string) ($b['status'] ?? ''));
        if ($statusComparison !== 0) {
          return $statusComparison;
        }

        return (int) ($b['updated'] ?? 0) <=> (int) ($a['updated'] ?? 0);
      });

      $result[$id] = ProgressData::fromDatabaseRow($rows[0]);
    }

    return $result;
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  #[\NoDiscard]
  public function getActiveConversions(): array {
    $query = $this->database->select(self::TABLE, 'p')
      ->fields('p')
      ->condition('status', [ConversionStatus::Pending->value, ConversionStatus::Converting->value], 'IN')
      ->orderBy('started', 'ASC');

    $statement = $query->execute();
    if ($statement === NULL) {
      return [];
    }

    $results = [];
    foreach ($statement as $row) {
      $results[] = ProgressData::fromDatabaseRow((array) $row);
    }

    return $results;
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  #[\NoDiscard]
  public function getStuckConversions(int $staleThreshold): array {
    $query = $this->database->select(self::TABLE, 'p')
      ->fields('p')
      ->condition('status', ConversionStatus::Converting->value)
      ->condition('updated', $staleThreshold, '<')
      ->orderBy('started', 'ASC');

    $statement = $query->execute();
    if ($statement === NULL) {
      return [];
    }

    $results = [];
    foreach ($statement as $row) {
      $results[] = ProgressData::fromDatabaseRow((array) $row);
    }

    return $results;
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  #[\NoDiscard]
  public function getActiveCount(): int {
    $now = microtime(TRUE);
    if ($this->cachedActiveCount !== NULL
      && ($now - $this->activeCountCachedAt) < $this->getActiveCountCacheTtl()) {
      return $this->cachedActiveCount;
    }

    $statement = $this->database->select(self::TABLE, 'p')
      ->condition('status', ConversionStatus::Converting->value)
      ->countQuery()
      ->execute();

    $this->cachedActiveCount = (int) ($statement?->fetchField() ?? 0);
    $this->activeCountCachedAt = $now;

    return $this->cachedActiveCount;
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  public function deleteProgress(int $mediaId, string $fieldName = ''): void {
    $delete = $this->database->delete(self::TABLE)
      ->condition('media_id', $mediaId);

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

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

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

    $deleted = (int) $this->database->delete(self::TABLE)
      ->condition('status', [ConversionStatus::Completed->value, ConversionStatus::Failed->value], 'IN')
      ->condition('completed', $cutoff, '<')
      ->execute();

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

    return $deleted;
  }

  /**
   * {@inheritdoc}
   */
  #[\Override]
  public function cleanupStuck(int $stuckTimeout = 7200): int {
    $cutoff = time() - $stuckTimeout;
    $now = time();

    // Count stuck records first. Only Converting items can be stuck;
    // Pending items are simply waiting in queue and should not be
    // force-failed by stuck detection.
    $countQuery = $this->database->select(self::TABLE, 'p')
      ->condition('status', ConversionStatus::Converting->value)
      ->condition('updated', $cutoff, '<')
      ->countQuery();

    $statement = $countQuery->execute();
    $count = (int) ($statement?->fetchField() ?? 0);

    if ($count === 0) {
      return 0;
    }

    // Batch UPDATE all stuck records in a single query.
    $this->database->update(self::TABLE)
      ->fields([
        'status' => ConversionStatus::Failed->value,
        'error_message' => sprintf('Stuck process detected — no update for over %d seconds.', $stuckTimeout),
        'updated' => $now,
        'completed' => $now,
      ])
      ->condition('status', ConversionStatus::Converting->value)
      ->condition('updated', $cutoff, '<')
      ->execute();

    $this->invalidateLiveCaches();

    return $count;
  }

  /**
   * Determines whether a progress update should be written to the database.
   *
   * Status changes (e.g., pending to converting) are always written
   * immediately. Same-status progress updates are throttled based on the
   * configured interval using the DB-backed `last_write` column, which
   * ensures throttling works correctly across concurrent workers.
   *
   * @param int $mediaId
   *   The media entity ID.
   * @param \Drupal\ffmpeg_media\Enum\ConversionStatus $status
   *   The status being written.
   * @param string $fieldName
   *   Scoped media field name used for field-specific throttling.
   *
   * @return bool
   *   TRUE if the update should be written, FALSE to skip.
   */
  #[\NoDiscard]
  private function shouldWriteProgress(int $mediaId, ConversionStatus $status, string $fieldName = ''): bool {
    // Terminal/transition statuses always write immediately.
    if (in_array($status, self::STATUS_TRANSITION_VALUES, TRUE)) {
      return TRUE;
    }

    $throttleInterval = $this->getProgressThrottleInterval();
    if ($throttleInterval <= 0) {
      return TRUE;
    }

    // Read last_write and status from DB for cross-worker throttling.
    $row = $this->getThrottleState($mediaId, $fieldName);

    if ($row === NULL) {
      // No DB row yet (before startTracking); allow write.
      return TRUE;
    }

    // Status change from last written status always writes immediately.
    if ($row['status'] !== $status->value) {
      return TRUE;
    }

    // No previous write recorded; allow first write.
    if ($row['last_write'] === NULL) {
      return TRUE;
    }

    // Throttle: skip if within the configured interval.
    $elapsed = time() - $row['last_write'];

    return $elapsed >= $throttleInterval;
  }

  /**
   * Reads throttle state from the database for a media_id.
   *
   * @param int $mediaId
   *   The media entity ID.
   * @param string $fieldName
   *   Scoped media field name used to load the throttled progress row.
   *
   * @return array{status: string, last_write: int|null}|null
   *   The current status and last_write timestamp, or NULL if no row.
   */
  #[\NoDiscard]
  private function getThrottleState(int $mediaId, string $fieldName = ''): ?array {
    $cacheKey = $mediaId . ':' . $fieldName;

    if (array_key_exists($cacheKey, $this->throttleStateCache)) {
      return $this->throttleStateCache[$cacheKey];
    }

    $row = $this->getProgressRow($mediaId, $fieldName);
    if ($row === NULL) {
      $this->throttleStateCache[$cacheKey] = NULL;
      return NULL;
    }

    $state = [
      'status' => (string) $row['status'],
      'last_write' => isset($row['last_write']) ? (int) $row['last_write'] : NULL,
    ];

    $this->throttleStateCache[$cacheKey] = $state;
    return $state;
  }

  /**
   * Resolves the field name that should be written for a progress update.
   */
  #[\NoDiscard]
  private function getWriteTargetFieldName(int $mediaId, string $fieldName): string {
    if ($fieldName === '') {
      return '';
    }

    $statement = $this->database->select(self::TABLE, 'p')
      ->fields('p', ['field_name'])
      ->condition('media_id', $mediaId)
      ->condition('field_name', [$fieldName, ''], 'IN')
      ->execute();

    if ($statement === NULL) {
      return $fieldName;
    }

    $rows = $statement->fetchCol();
    if (in_array($fieldName, $rows, TRUE)) {
      return $fieldName;
    }

    return in_array('', $rows, TRUE) ? '' : $fieldName;
  }

  /**
   * Returns a scoped progress row with legacy fallback.
   *
   * @return array<string, int|string|null>|null
   *   Database row or NULL.
   */
  #[\NoDiscard]
  private function getProgressRow(int $mediaId, string $fieldName = ''): ?array {
    if ($fieldName !== '') {
      $statement = $this->database->select(self::TABLE, 'p')
        ->fields('p')
        ->condition('media_id', $mediaId)
        ->condition('field_name', [$fieldName, ''], 'IN')
        ->execute();

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

      $rows = $statement->fetchAllAssoc('field_name');
      if ($rows === [] || $rows === FALSE) {
        return NULL;
      }
      if (isset($rows[$fieldName])) {
        return (array) $rows[$fieldName];
      }
      if (isset($rows[''])) {
        return (array) $rows[''];
      }

      return NULL;
    }

    $statement = $this->database->select(self::TABLE, 'p')
      ->fields('p')
      ->condition('media_id', $mediaId)
      ->execute();

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

    $rawRows = $statement->fetchAll();
    if (!is_array($rawRows) || $rawRows === []) {
      return NULL;
    }

    $rows = array_map(
      static fn(array|object $row): array => (array) $row,
      $rawRows,
    );
    if ($rows === []) {
      return NULL;
    }

    usort($rows, static function (array $a, array $b): int {
      $priority = static function (string $status): int {
        return match ($status) {
          ConversionStatus::Converting->value => 0,
          ConversionStatus::Pending->value => 1,
          ConversionStatus::Failed->value => 2,
          ConversionStatus::Completed->value => 3,
          default => 4,
        };
      };

      $statusComparison = $priority((string) ($a['status'] ?? '')) <=> $priority((string) ($b['status'] ?? ''));
      if ($statusComparison !== 0) {
        return $statusComparison;
      }

      return (int) ($b['updated'] ?? 0) <=> (int) ($a['updated'] ?? 0);
    });

    return $rows[0];
  }

  /**
   * Gets the progress throttle interval from config or default.
   *
   * @return int
   *   Minimum seconds between DB writes for the same media_id.
   */
  #[\NoDiscard]
  private function getProgressThrottleInterval(): int {
    $config = $this->configFactory->get('ffmpeg_media.settings');
    $interval = $config->get('progress_throttle_interval');

    return is_int($interval) && $interval >= 0
      ? $interval
      : self::DEFAULT_PROGRESS_THROTTLE_INTERVAL;
  }

  /**
   * Gets the active count cache TTL from config or default.
   *
   * @return float
   *   Cache TTL in seconds.
   */
  #[\NoDiscard]
  private function getActiveCountCacheTtl(): float {
    $config = $this->configFactory->get('ffmpeg_media.settings');
    $ttl = $config->get('active_count_cache_ttl');

    return is_numeric($ttl) && (float) $ttl >= 0
      ? (float) $ttl
      : self::ACTIVE_COUNT_CACHE_TTL;
  }

  /**
   * Invalidates the in-memory active count cache.
   */
  private function invalidateActiveCountCache(): void {
    $this->cachedActiveCount = NULL;
    $this->activeCountCachedAt = 0.0;
  }

  /**
   * Invalidates dashboard, stats, and per-media output cache tags.
   *
   * @param int|null $mediaId
   *   Optional media entity ID. When provided, the formatter output cache
   *   tag 'ffmpeg_media:outputs:{mediaId}' is also invalidated so that
   *   ConvertedVideoFormatter picks up progress changes immediately.
   */
  private function invalidateLiveCaches(?int $mediaId = NULL): void {
    $tags = [
      'ffmpeg_media:stats',
      'ffmpeg_media:dashboard',
    ];
    if ($mediaId !== NULL) {
      $tags[] = 'ffmpeg_media:outputs:' . $mediaId;
    }
    $this->cacheTagsInvalidator?->invalidateTags($tags);
  }

}
