<?php

declare(strict_types=1);

namespace Drupal\acquia_dam;

use Drupal\acquia_dam\Client\AcquiaDamClientFactory;
use Drupal\acquia_dam\Entity\MediaSourceField;
use Drupal\Component\Datetime\TimeInterface;
use Drupal\Core\Datetime\DrupalDateTime;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Logger\LoggerChannelFactoryInterface;
use Drupal\Core\Queue\QueueFactory;
use Drupal\Core\Queue\QueueWorkerManagerInterface;
use Drupal\Core\StringTranslation\StringTranslationTrait;
use Drupal\media\MediaInterface;

/**
 * Service for processing Acquia DAM asset updates.
 */
final class AssetUpdater {

  use StringTranslationTrait;

  /**
   * Number of items per API search page.
   */
  const SEARCH_PAGE_LIMIT = 100;

  /**
   * Number of items per processing chunk.
   */
  const CHUNK_SIZE = 50;

  /**
   * Constructs a new AssetUpdater.
   *
   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entityTypeManager
   *   The entity type manager.
   * @param \Drupal\acquia_dam\AssetUpdateChecker $assetUpdateChecker
   *   DAM asset update checker.
   * @param \Drupal\acquia_dam\AssetVersionResolver $assetVersionResolver
   *   DAM asset version resolver.
   * @param \Drupal\Core\Queue\QueueFactory $queueFactory
   *   The queue factory.
   * @param \Drupal\Core\Queue\QueueWorkerManagerInterface $queueManager
   *   The queue worker manager.
   * @param \Drupal\Core\Logger\LoggerChannelFactoryInterface $loggerFactory
   *   The logger factory.
   * @param \Drupal\acquia_dam\Client\AcquiaDamClientFactory $clientFactory
   *   The Acquia DAM client factory.
   * @param \Drupal\Component\Datetime\TimeInterface $time
   *   The time service.
   */
  public function __construct(
    private readonly EntityTypeManagerInterface $entityTypeManager,
    private readonly AssetUpdateChecker $assetUpdateChecker,
    private readonly AssetVersionResolver $assetVersionResolver,
    private readonly QueueFactory $queueFactory,
    private readonly QueueWorkerManagerInterface $queueManager,
    private readonly LoggerChannelFactoryInterface $loggerFactory,
    private readonly AcquiaDamClientFactory $clientFactory,
    private readonly TimeInterface $time,
  ) {}

  /**
   * Sync assets from DAM with optional finalized version check.
   *
   * This is the main orchestration method that:
   * 1. Fetches assets from DAM (optionally filtered by date)
   * 2. Categorizes into unpublish/update groups
   * 3. Processes unpublish operations
   * 4. Queues update operations
   * 5. Optionally checks remaining media for finalized version changes
   *
   * @param array $options
   *   Options array containing:
   *   - since-timestamp: int|null - Filter by date (null or 0 = all assets)
   *   - finalized-version: bool - Check remaining media for version changes
   *   - batch-size: int - Processing chunk size
   *   - media-type: string|null - Filter by media type
   *
   * @return array
   *   Summary array with processing statistics.
   *
   * @throws \Drupal\Component\Plugin\Exception\InvalidPluginDefinitionException
   * @throws \Drupal\Component\Plugin\Exception\PluginNotFoundException
   * @throws \Drupal\acquia_dam\Exception\DamClientException
   */
  public function syncAssets(array $options = []): array {
    $since_timestamp = $options['since-timestamp'] ?? NULL;
    $date_filter = $options['date-filter'] ?? NULL;
    $check_finalized = $options['finalized-version'] ?? FALSE;
    $batch_size = $options['batch-size'] ?? self::CHUNK_SIZE;
    $media_type_filter = $options['media-type'] ?? NULL;

    // Step 1: Build query.
    if ($since_timestamp === NULL) {
      $query = "lastEditDate:[$date_filter]";
    }
    else {
      $query = $this->generateQuery($since_timestamp);
    }

    // Step 2: Fetch from DAM page by page and categorize immediately.
    //
    // fetchUpdatedAssets() is a Generator that yields one scroll page at a
    // time. Calling categorizeAssets() per page means only one page of raw
    // API data (~SEARCH_PAGE_LIMIT items × ~2–5 KB each ≈ 200–500 KB) lives
    // in memory at once. The accumulated maps below are tiny
    // ([media_id => asset_id], ~70 bytes/entry), so memory stays bounded even
    // for libraries with tens of thousands of assets.
    $all_unpublish = [];
    $all_update = [];
    $total_count = 0;

    foreach ($this->fetchUpdatedAssets($query) as $page_items) {
      $total_count += count($page_items);
      $categorized_page = $this->categorizeAssets($page_items);
      // Use the union operator: media IDs are unique across pages so keys
      // never collide, and unlike array_merge() it preserves integer keys.
      $all_unpublish += $categorized_page['unpublish'];
      $all_update += $categorized_page['update'];
    }

    // Initialize stats.
    $stats = [
      'dam_processed' => $total_count,
      'unpublished' => 0,
      'queued_from_dam' => 0,
      'queued_from_version_check' => 0,
    ];

    if ($total_count === 0) {
      // If --finalized-version is set, still check all media.
      if ($check_finalized) {
        $version_check_stats = $this->checkRemainingMediaForVersionUpdates(
          [],
          $media_type_filter,
          $batch_size
        );
        $stats['queued_from_version_check'] = $version_check_stats['queued'];
      }
      return $stats;
    }

    $processed_media = [];

    // Step 3: Unpublish immediately.
    if (!empty($all_unpublish)) {
      $this->unpublishAssets($all_unpublish);
      $processed_media = array_merge($processed_media, array_keys($all_unpublish));
      $stats['unpublished'] = count($all_unpublish);
    }

    // Step 4: Queue DAM-reported updates.
    $update_map = $all_update;

    // Apply media-type filter if specified.
    if ($media_type_filter && !empty($update_map)) {
      $update_map = $this->filterByMediaType($update_map, $media_type_filter);
    }

    if (!empty($update_map)) {
      $this->queueUpdateAssets($update_map);
      $processed_media = array_merge($processed_media, array_keys($update_map));
      $stats['queued_from_dam'] = count($update_map);
    }

    // Step 5: If --finalized-version flag is set, check remaining media.
    if ($check_finalized) {
      $version_check_stats = $this->checkRemainingMediaForVersionUpdates(
        $processed_media,
        $media_type_filter,
        $batch_size
      );
      $stats['queued_from_version_check'] = $version_check_stats['queued'];
    }

    return $stats;
  }

  /**
   * Yields updated asset pages from the DAM scroll search API.
   *
   * Each yielded value is one page of raw asset items (up to
   * SEARCH_PAGE_LIMIT entries). Only one page of API data lives in memory
   * at a time; callers must not accumulate yields into a flat array.
   *
   * If a scroll continuation request fails (e.g. the scroll session times
   * out), the exception is caught here, a warning is logged, and the
   * generator returns. Items already yielded from earlier pages are not
   * affected — callers will have already processed them.
   *
   * @param string $query
   *   The search query string (e.g. "lastEditDate:[after …]").
   *
   * @yields array
   *   One page of raw asset items from the DAM scroll API response.
   */
  public function fetchUpdatedAssets(string $query): \Generator {
    $client = $this->clientFactory->getSiteClient();

    try {
      // yield from delegates every page yielded by scrollSearch() directly to
      // this generator's caller, without buffering the full result set.
      yield from $client->scrollSearch(
        $query,
        '-created_date',
        self::SEARCH_PAGE_LIMIT,
        ['file_properties'],
        TRUE
      );
    }
    catch (\Exception $exception) {
      $this->loggerFactory->get('acquia_dam.media_item_update')->warning(
        'Unable to get asset list from API. Error: %message',
        ['%message' => $exception->getMessage()]
      );
    }
  }

  /**
   * Categorizes active DAM assets into two processing groups.
   *
   * Uses the search response data directly to classify assets, eliminating
   * the per-asset getAsset() and getFinalizedVersion() API calls that the
   * previous checkAssets() path required:
   *  - 'unpublish': deleted, expired, or unreleased assets — acted on
   *    immediately without queuing.
   *  - 'update': released assets that need metadata / version sync via the
   *    MediaItemUpdater queue worker.
   *
   * @param array $items
   *   Raw item list from the DAM search API response.
   *
   * @return array
   *   Associative array with two keys:
   *   - 'unpublish': [media_id => asset_id]
   *   - 'update':    [media_id => asset_id]
   *
   * @throws \Drupal\Component\Plugin\Exception\InvalidPluginDefinitionException
   * @throws \Drupal\Component\Plugin\Exception\PluginNotFoundException
   */
  public function categorizeAssets(array $items): array {
    if (count($items) === 0) {
      return ['unpublish' => [], 'update' => []];
    }

    // Build a lookup map of asset_id => search_item for O(1) classification.
    $items_by_uuid = array_column($items, NULL, 'id');

    $media_storage = $this->entityTypeManager->getStorage('media');
    $media_ids = $media_storage
      ->getQuery()
      ->accessCheck(FALSE)
      ->condition('acquia_dam_asset_id', array_keys($items_by_uuid), 'IN')
      ->execute();

    $unpublish = [];
    $update    = [];

    foreach ($media_storage->loadMultiple($media_ids) as $media_id => $media) {
      $asset_id    = $media->get(MediaSourceField::SOURCE_FIELD_NAME)->asset_id;
      $search_item = $items_by_uuid[$asset_id] ?? NULL;

      if ($search_item === NULL) {
        continue;
      }

      // Classify as "unpublish" when:
      // (a) deleted_date is set — the asset was removed from DAM, or
      // (b) released_and_not_expired is false — asset is expired / unreleased.
      if (!empty($search_item['deleted_date']) || empty($search_item['released_and_not_expired'])) {
        $unpublish[(int) $media_id] = $asset_id;
      }
      else {
        $update[(int) $media_id] = $asset_id;
      }
    }

    return ['unpublish' => $unpublish, 'update' => $update];
  }

  /**
   * Unpublish media items for deleted, expired, or unreleased assets.
   *
   * @param array $unpublish_map
   *   Asset IDs to unpublish, keyed by media ID: [media_id => asset_id].
   *
   * @throws \Drupal\Component\Plugin\Exception\InvalidPluginDefinitionException
   * @throws \Drupal\Component\Plugin\Exception\PluginNotFoundException
   * @throws \Drupal\Core\Entity\EntityStorageException
   */
  public function unpublishAssets(array $unpublish_map): void {
    $client = $this->clientFactory->getSiteClient();
    $media_storage = $this->entityTypeManager->getStorage('media');

    foreach (array_chunk($unpublish_map, self::CHUNK_SIZE, TRUE) as $chunk) {
      $asset_ids = array_values($chunk);
      $media_ids = array_keys($chunk);

      $client->deleteCachedAssets($asset_ids);

      foreach ($media_storage->loadMultiple($media_ids) as $media_item) {
        assert($media_item instanceof MediaInterface);
        if (!$media_item->isPublished()) {
          continue;
        }
        $media_item->setNewRevision();
        $media_item->setUnpublished()
          ->setRevisionCreationTime($this->time->getCurrentTime())
          ->setRevisionLogMessage($this->t('Automatically unpublished as the associated DAM asset is unavailable.'))
          ->setRevisionUserId(0)
          ->save();
      }
    }
  }

  /**
   * Enqueue assets that require metadata/file-property sync.
   *
   * @param array $update_map
   *   Asset IDs to update, keyed by media ID: [media_id => asset_id].
   *
   * @throws \Drupal\Component\Plugin\Exception\InvalidPluginDefinitionException
   * @throws \Drupal\Component\Plugin\Exception\PluginNotFoundException
   */
  public function queueUpdateAssets(array $update_map): void {
    $client = $this->clientFactory->getSiteClient();
    $media_storage = $this->entityTypeManager->getStorage('media');
    $queue = $this->queueFactory->get('acquia_dam_media_item_update');

    foreach (array_chunk($update_map, self::CHUNK_SIZE, TRUE) as $chunk) {
      $asset_ids = array_values($chunk);
      $media_ids = array_keys($chunk);

      $client->deleteCachedAssets($asset_ids);

      foreach ($media_storage->loadMultiple($media_ids) as $media_item) {
        assert($media_item instanceof MediaInterface);
        $asset_field = $media_item->get(MediaSourceField::SOURCE_FIELD_NAME)->first();
        if (!$asset_field) {
          continue;
        }
        $asset_field_values = $asset_field->getValue();
        // There is no user in the cron/drush context; send the Anonymous UID.
        $queue->createItem([
          'asset_id' => $asset_field_values['asset_id'],
          'media_id' => $media_item->id(),
          'user_id'  => 0,
        ]);
      }
    }
  }

  /**
   * Check media items not processed by DAM sync for finalized version changes.
   *
   * This handles the case where an asset's finalized version changes
   * but DAM doesn't report it in the lastEditDate query (edge case).
   *
   * @param array $processed_media
   *   Media IDs already handled by DAM sync.
   * @param string|null $media_type_filter
   *   Optional media type filter.
   * @param int $batch_size
   *   Processing chunk size.
   *
   * @return array
   *   Statistics: ['checked' => int, 'queued' => int]
   *
   * @throws \Drupal\Component\Plugin\Exception\InvalidPluginDefinitionException
   * @throws \Drupal\Component\Plugin\Exception\PluginNotFoundException
   */
  protected function checkRemainingMediaForVersionUpdates(
    array $processed_media,
    ?string $media_type_filter,
    int $batch_size
  ): array {
    $media_storage = $this->entityTypeManager->getStorage('media');

    // Build query for remaining media.
    $query = $media_storage->getQuery()
      ->accessCheck(FALSE)
      ->condition(MediaSourceField::SOURCE_FIELD_NAME . '.asset_id', NULL, 'IS NOT NULL');

    // Exclude already processed.
    if (!empty($processed_media)) {
      $query->condition('mid', $processed_media, 'NOT IN');
    }

    // Apply media type filter.
    if ($media_type_filter) {
      $query->condition('bundle', $media_type_filter);
    }

    $remaining_ids = $query->execute();

    if (empty($remaining_ids)) {
      return ['checked' => 0, 'queued' => 0];
    }

    $checked = 0;
    $queued = 0;

    // Process in chunks.
    foreach (array_chunk($remaining_ids, $batch_size, TRUE) as $chunk) {
      $media_items = $media_storage->loadMultiple($chunk);

      foreach ($media_items as $media_item) {
        $checked++;

        // Check if finalized version differs and queue if needed.
        // Returns FALSE if queued, TRUE if up-to-date.
        $is_current = $this->checkAssetsFinalizedVersion($media_item, 0);

        if ($is_current === FALSE) {
          $queued++;
        }
      }

      // Clear cache for memory management.
      $media_storage->resetCache($chunk);
    }

    return ['checked' => $checked, 'queued' => $queued];
  }

  /**
   * Filter update map by media type.
   *
   * @param array $update_map
   *   [media_id => asset_id] map.
   * @param string $media_type
   *   Media type to filter by.
   *
   * @return array
   *   Filtered map.
   *
   * @throws \Drupal\Component\Plugin\Exception\InvalidPluginDefinitionException
   * @throws \Drupal\Component\Plugin\Exception\PluginNotFoundException
   */
  protected function filterByMediaType(array $update_map, string $media_type): array {
    if (empty($update_map)) {
      return [];
    }

    $media_storage = $this->entityTypeManager->getStorage('media');
    $media_ids = $media_storage
      ->getQuery()
      ->accessCheck(FALSE)
      ->condition('mid', array_keys($update_map), 'IN')
      ->condition('bundle', $media_type)
      ->execute();

    return array_intersect_key($update_map, array_flip($media_ids));
  }

  /**
   * Process the Acquia DAM asset update queue.
   *
   * @param array $options
   *   Options array containing:
   *   - limit: Maximum number of queue items to process (0 = no limit)
   *
   * @return array
   *   Summary array with processing results.
   */
  public function processQueue(array $options = []): array {
    $limit = (int) ($options['limit'] ?? 0);

    $queue = $this->queueFactory->get('acquia_dam_media_item_update');
    $queue_worker = $this->queueManager->createInstance('acquia_dam_media_item_update');

    $processed = 0;
    $errors = 0;
    $initial_count = $queue->numberOfItems();

    while (($limit == 0 || $processed < $limit) && ($item = $queue->claimItem())) {
      try {
        $queue_worker->processItem($item->data);
        $queue->deleteItem($item);
        $processed++;
      }
      catch (\Exception $e) {
        $errors++;
        $this->loggerFactory->get('acquia_dam')->error(
          'Error processing queue item: @message',
          ['@message' => $e->getMessage()]
        );
        // Release the item back to the queue on error.
        $queue->releaseItem($item);
      }
    }

    return [
      'initial_count' => $initial_count,
      'processed' => $processed,
      'errors' => $errors,
      'remaining' => $queue->numberOfItems(),
      'message' => 'Queue processing completed.',
    ];
  }

  /**
   * Get all media types that use DAM assets.
   *
   * @param string|null $media_type_filter
   *   Optional filter for a specific media type.
   *
   * @return array
   *   Array of media type IDs that use DAM assets.
   *
   * @throws \Drupal\Component\Plugin\Exception\InvalidPluginDefinitionException
   * @throws \Drupal\Component\Plugin\Exception\PluginNotFoundException
   */
  public function getDamMediaTypes(?string $media_type_filter = NULL): array {
    $media_type_storage = $this->entityTypeManager->getStorage('media_type');

    // Find all media types that use DAM assets.
    $media_types = $media_type_storage->loadByProperties([
      'source_configuration.source_field' => MediaSourceField::SOURCE_FIELD_NAME,
    ]);

    // Extract media type IDs using array_map.
    $media_types = array_map(fn($media_type) => $media_type->id(), $media_types);

    // Return filtered media type or all media types.
    return $media_type_filter && isset($media_types[$media_type_filter])
      ? [$media_type_filter => $media_type_filter]
      : $media_types;
  }

  /**
   * Check if the asset is finalized version.
   *
   * @param \Drupal\media\MediaInterface $media_item
   *   The media item to check.
   * @param int $user_id
   *   The user id.
   *
   * @return bool
   *   TRUE if the asset is finalized, FALSE otherwise.
   */
  public function checkAssetsFinalizedVersion(MediaInterface $media_item, int $user_id = 0): bool {
    assert($media_item instanceof MediaInterface);
    $asset_field = $media_item->get(MediaSourceField::SOURCE_FIELD_NAME)->first();
    if (!$asset_field || is_null($asset_field->getValue()['asset_id'])) {
      $this->loggerFactory->get('acquia_dam')->error($this->t('Media item with ID %media_item_id has no asset ID set, which normally should not happen.', [
        '%media_item_id' => $media_item->id(),
      ]), ['link' => $media_item->toLink($this->t('View'))->toString()]);
      return FALSE;
    }

    $asset_ids = $asset_field->getValue();
    $asset_finalized_version_id = $this->assetVersionResolver->getFinalizedVersion($asset_ids['asset_id']);

    // If finalized version empty means asset deleted in DAM, unpublish media.
    if (empty($asset_finalized_version_id)) {
      $media_item->setNewRevision();
      $media_item->setUnpublished()
        ->setRevisionCreationTime($this->time->getCurrentTime())
        ->setRevisionLogMessage($this->t('Automatically unpublished as the associated DAM asset is unavailable.'))
        ->setRevisionUserId(0)
        ->save();
      return TRUE;
    }
    // If finalized version matched do nothing.
    elseif ($asset_ids['version_id'] === $asset_finalized_version_id) {
      return TRUE;
    }
    // If finalized version do not matches add to queue.
    else {
      // Now if updating is needed, then queue the task for later processing.
      $this->queueFactory->get('acquia_dam_media_item_update')->createItem([
        'asset_id' => $asset_ids['asset_id'],
        'media_id' => $media_item->id(),
        'user_id' => $user_id,
      ]);
      return FALSE;
    }
  }

  /**
   * The helper function to generate query.
   *
   * @param int|null $since_timestamp
   *   Unix timestamp to filter by lastEditDate. NULL or 0 = no filter (all).
   *
   * @return string
   *   The generated query string.
   */
  private function generateQuery(?int $since_timestamp): string {
    // Build query - use old date for "all assets" case.
    if (empty($since_timestamp)) {
      $formatted = DrupalDateTime::createFromTimestamp(0, 'UTC')
        ->format('Y-m-d\TH:i:s\Z');
    }
    else {
      $formatted = DrupalDateTime::createFromTimestamp($since_timestamp, 'UTC')
        ->format('Y-m-d\TH:i:s\Z');
    }

    return "lastEditDate:[after $formatted]";
  }

}
