<?php

declare(strict_types=1);

namespace Drupal\acquia_dam\Drush\Commands;

use Drupal\acquia_dam\AssetFileEntityHelper;
use Drupal\acquia_dam\AssetUpdater;
use Drupal\acquia_dam\Client\AcquiaDamClient;
use Drupal\acquia_dam\Client\AcquiaDamClientFactory;
use Drupal\acquia_dam\MediaTypeResolver;
use Drupal\Core\Database\Connection;
use Drupal\Core\Entity\EntityStorageInterface;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\StringTranslation\StringTranslationTrait;
use Drupal\media\MediaInterface;
use Drush\Attributes as CLI;
use Drush\Commands\DrushCommands;
use Symfony\Component\DependencyInjection\ContainerInterface;

/**
 * Drush commands for Acquia DAM media item operations.
 */
final class AssetUpdateDrushCommands extends DrushCommands {

  use StringTranslationTrait;

  /**
   * Media image field table name.
   */
  private const MEDIA_IMAGE_TABLE = 'media__acquia_dam_managed_image';

  /**
   * Media file field table name.
   */
  private const MEDIA_FILE_TABLE = 'media__acquia_dam_managed_file';

  /**
   * Constructs a new AssetUpdateDrushCommands object.
   *
   * @param \Drupal\acquia_dam\AssetUpdater $assetUpdater
   *   The asset update processor service.
   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entityTypeManager
   *   The entity type manager.
   * @param \Drupal\acquia_dam\Client\AcquiaDamClientFactory $acquiaDamClientFactory
   *   The Acquia DAM client factory.
   * @param \Drupal\acquia_dam\MediaTypeResolver $mediaTypeResolver
   *   The media type resolver service.
   * @param \Drupal\acquia_dam\AssetFileEntityHelper $assetFileHelper
   *   The asset file entity helper service.
   * @param \Drupal\Core\Database\Connection $connection
   *   The database connection.
   */
  public function __construct(
    private readonly AssetUpdater $assetUpdater,
    private readonly EntityTypeManagerInterface $entityTypeManager,
    private readonly AcquiaDamClientFactory $acquiaDamClientFactory,
    private readonly MediaTypeResolver $mediaTypeResolver,
    private readonly AssetFileEntityHelper $assetFileHelper,
    private readonly Connection $connection,
  ) {
    parent::__construct();
  }

  /**
   * {@inheritdoc}
   */
  public static function create(ContainerInterface $container): self {
    return new static(
      $container->get('acquia_dam.asset_updater'),
      $container->get('entity_type.manager'),
      $container->get('acquia_dam.client.factory'),
      $container->get('acquia_dam.media_type_resolver'),
      $container->get('acquia_dam.asset_file_helper'),
      $container->get('database')
    );
  }

  /**
   * Queue DAM assets for update based on changes in Acquia DAM.
   *
   * This command fetches assets from DAM that have been modified and queues
   * them for update. Optionally performs additional finalized version checks
   * on media items not reported as changed by DAM.
   *
   * @param array $options
   *   An associative array of options whose values come from cli, aliases,
   *   config, etc.
   *
   * @command acquia-dam:queue-update-assets
   * @aliases adqua
   *
   * @option date-filter
   *   Filter assets by date. Formats: "mm/dd/yyyy to mm/dd/yyyy" (between),
   *   "after mm/dd/yyyy" (after a date), "before mm/dd/yyyy" (before a date).
   * @option finalized-version
   *   Additionally check all media items not in DAM's change list for
   *   finalized version mismatches and queue if needed.
   * @option batch-size
   *   Number of media items to process in each batch. Defaults to 50.
   * @option media-type
   *   Specific media type to process.
   *   If not specified, all DAM media types will be processed.
   *
   * @usage acquia-dam:queue-update-assets
   *   Queue all assets from DAM.
   * @usage acquia-dam:queue-update-assets --date-filter="06/15/2026 to 07/15/2026"
   *   Queue only assets changed between the given dates.
   * @usage acquia-dam:queue-update-assets --date-filter="after 06/15/2026"
   *   Queue only assets changed after the given date.
   * @usage acquia-dam:queue-update-assets --date-filter="before 06/15/2026"
   *   Queue only assets changed before the given date.
   * @usage acquia-dam:queue-update-assets --finalized-version
   *   Queue all DAM changes PLUS check finalized versions for remaining media.
   * @usage acquia-dam:queue-update-assets --date-filter="after 06/15/2026" --finalized-version
   *   Queue recent changes AND check finalized versions for all other media.
   * @usage acquia-dam:queue-update-assets --batch-size=100
   *   Queue in batches of 100 items.
   * @usage acquia-dam:queue-update-assets --media-type=acquia_dam_image_asset
   *   Queue only media items of type acquia_dam_image_asset.
   *
   * @help This command uses the DAM API to efficiently fetch and process
   *   asset changes. When --finalized-version is enabled, it performs an
   *   additional check on media items not reported by DAM to catch edge cases
   *   where the finalized version changes without triggering a lastEditDate
   *   update in DAM. Use --date-filter to limit assets by date. Supported
   *   formats: "mm/dd/yyyy to mm/dd/yyyy" (between), "after mm/dd/yyyy",
   *   or "before mm/dd/yyyy".
   */

  #[CLI\Command(name: 'acquia-dam:queue-update-assets', aliases: ['adqua'])]
  #[CLI\Option(name: 'date-filter', description: 'Filter by date. Formats: "mm/dd/yyyy to mm/dd/yyyy" (between), "after mm/dd/yyyy", "before mm/dd/yyyy".')]
  #[CLI\Option(name: 'finalized-version', description: 'Additionally check remaining media for finalized version changes')]
  #[CLI\Option(name: 'batch-size', description: 'Number of media items to process in each batch')]
  #[CLI\Option(name: 'media-type', description: 'Specific media type to process')]
  #[CLI\Usage(name: 'acquia-dam:queue-update-assets', description: 'Queue all assets from DAM')]
  #[CLI\Usage(name: 'acquia-dam:queue-update-assets --date-filter="06/15/2026 to 07/15/2026"', description: 'Queue only assets changed between the given dates')]
  #[CLI\Usage(name: 'acquia-dam:queue-update-assets --date-filter="after 06/15/2026"', description: 'Queue only assets changed after the given date')]
  #[CLI\Usage(name: 'acquia-dam:queue-update-assets --date-filter="before 06/15/2026"', description: 'Queue only assets changed before the given date')]
  #[CLI\Usage(name: 'acquia-dam:queue-update-assets --finalized-version', description: 'Queue DAM changes + check finalized versions')]
  #[CLI\Usage(name: 'acquia-dam:queue-update-assets --batch-size=100', description: 'Queue in batches of 100')]
  #[CLI\Usage(name: 'acquia-dam:queue-update-assets --media-type=acquia_dam_image_asset', description: 'Queue only specific media type')]
  #[CLI\Help(description: 'This command uses the DAM API to efficiently fetch
  and process asset changes. When --finalized-version is enabled, it performs
  an additional check on media items not reported by DAM to catch edge cases.
  Use --date-filter to limit assets by date. Supported formats:
  "mm/dd/yyyy to mm/dd/yyyy" (between), "after mm/dd/yyyy", "before mm/dd/yyyy".')]
  public function queueAssets(
    array $options = [
      'date-filter' => '',
      'finalized-version' => FALSE,
      'batch-size' => 50,
      'media-type' => NULL,
    ],
  ): void {
    try {
      $this->output()->writeln('<info>=== STARTING DAM ASSET QUEUE OPERATION ===</info>');
      if (!empty($options['date-filter'])) {
        $this->output()->writeln(sprintf('<comment>Date filter applied: %s</comment>', $options['date-filter']));
      }
      else {
        $this->output()->writeln('<comment>Processing: ALL assets from DAM</comment>');
      }

      if ($options['finalized-version']) {
        $this->output()->writeln('<comment>Additional check: Finalized version verification enabled</comment>');
      }

      $this->output()->writeln('');

      // Use the new unified syncAssets method.
      $result = $this->assetUpdater->syncAssets([
        'since-timestamp' => NULL,
        'date-filter' => !empty($options['date-filter']) ? (string) $options['date-filter'] : 'before ' . date('m/d/Y'),
        'finalized-version' => (bool) $options['finalized-version'],
        'batch-size' => (int) $options['batch-size'],
        'media-type' => $options['media-type'],
      ]);

      // Display results.
      $this->output()->writeln('<info>=== DAM API SYNC RESULTS ===</info>');
      $this->output()->writeln(sprintf('Assets reported by DAM: <comment>%d</comment>',
        $result['dam_processed']));
      $this->output()->writeln(sprintf('Media items unpublished: <comment>%d</comment>',
        $result['unpublished']));
      $this->output()->writeln(sprintf('Media items queued from DAM changes: <comment>%d</comment>',
        $result['queued_from_dam']));

      if ($options['finalized-version']) {
        $this->output()->writeln('');
        $this->output()->writeln('<info>=== FINALIZED VERSION CHECK ===</info>');
        $this->output()->writeln(sprintf('Additional media queued from version check: <comment>%d</comment>',
          $result['queued_from_version_check']));
      }

      $total_queued = $result['queued_from_dam'] + $result['queued_from_version_check'];
      $this->output()->writeln('');
      $this->output()->writeln(sprintf('<info>TOTAL media items queued for update: %d</info>', $total_queued));

      if ($total_queued > 0) {
        $this->output()->writeln('<info>Updates have been queued and will be processed during the next cron run or when the queue is manually processed.</info>');
      }
      else {
        $this->output()->writeln('<info>No updates needed - all media items are up to date.</info>');
      }

    }
    catch (\Exception $e) {
      $this->output()->writeln(sprintf('<error>An error occurred: %s</error>', $e->getMessage()));
      $this->logger()->error('Error in DAM queue command: @message', ['@message' => $e->getMessage()]);
    }
  }

  /**
   * Process the Acquia DAM asset update queue.
   *
   * This command processes the queued media item updates immediately instead
   * of waiting for cron.
   *
   * @param array $options
   *   An associative array of options whose values come from cli, aliases,
   *    config, etc.
   *
   * @command acquia-dam:process-update-queue
   * @aliases adpuq
   *
   * @option limit
   *   Maximum number of queue items to process. Default to 0 (no limit).
   * @usage acquia-dam:process-update-queue
   *   Process all queued DAM media updates.
   * @usage acquia-dam:process-update-queue --limit=100
   *   Process up to 100 queued DAM media updates.
   * @help This command processes the DAM media update queue and provides
   *   detailed progress reporting, including a summary of processed, failed,
   *   and remaining items. Use the --limit option to control the number of
   *   items processed in a single execution.
   */
  #[CLI\Command(name: 'acquia-dam:process-update-queue', aliases: ['adpuq'])]
  #[CLI\Option(name: 'limit', description: 'Maximum number of queue items to process. Defaults to 0 (no limit).')]
  #[CLI\Usage(name: 'acquia-dam:process-update-queue', description: 'Process all queued DAM media updates.')]
  #[CLI\Usage(name: 'acquia-dam:process-update-queue --limit=100', description: 'Process only the first 100 items from the queue.')]
  #[CLI\Help(description: 'This command processes the DAM media update queue and
  provides detailed progress reporting, including a summary of processed,
  failed, and remaining items. Use the --limit option to control the number
  of items processed in a single execution.')]
  public function processQueue(array $options = ['limit' => 0]): void {
    $result = $this->assetUpdater->processQueue($options);

    $this->output()->writeln(sprintf('<info>Processing DAM media update queue (items in queue: %d)</info>', $result['initial_count']));

    // Show progress during processing.
    if ($result['processed'] > 0) {
      for ($i = 10; $i <= $result['processed']; $i += 10) {
        if ($i <= $result['processed']) {
          $this->output()->writeln(sprintf('<comment>Processed %d items...</comment>', min($i, $result['processed'])));
        }
      }
    }

    $this->output()->writeln('');
    $this->output()->writeln('<info>=== QUEUE PROCESSING SUMMARY ===</info>');
    $this->output()->writeln(sprintf('Items processed: <comment>%d</comment>', $result['processed']));
    $this->output()->writeln(sprintf('Remaining items in queue: <comment>%d</comment>', $result['remaining']));

    if ($result['errors'] > 0) {
      $this->output()->writeln(sprintf('Errors encountered: <error>%d</error>', $result['errors']));
    }
  }

  /**
   * Update Acquia DAM assets (queue and process in one command).
   *
   * This command first queues all DAM media items for update, then immediately
   * processes the queue instead of waiting for cron.
   *
   * @param array $options
   *   An associative array of options whose values come from cli, aliases,
   *    config, etc.
   *
   * @command acquia-dam:update-assets
   * @alias adua
   *
   * @option date-filter
   *   Filter assets by date. Formats: "mm/dd/yyyy to mm/dd/yyyy" (between),
   *   "after mm/dd/yyyy" (after a date), "before mm/dd/yyyy" (before a date).
   * @option finalized-version
   *   Additionally check all media items not in DAM's change list for
   *   finalized version mismatches and queue if needed.
   * @option batch-size
   *   Number of media items to process in each batch. Defaults to 50.
   * @option limit
   *   Maximum number of queue items to process. Defaults to 0 (no limit).
   * @option media-type
   *   Specific media type to process.
   *   If not specified, all DAM media types will be processed.
   *
   * @usage acquia-dam:update-assets
   *   Queue and process all DAM media updates.
   * @usage acquia-dam:update-assets --date-filter="06/15/2026 to 07/15/2026"
   *   Queue and process only assets changed between the given dates.
   * @usage acquia-dam:update-assets --date-filter="after 06/15/2026"
   *   Queue and process only assets changed after the given date.
   * @usage acquia-dam:update-assets --date-filter="before 06/15/2026"
   *   Queue and process only assets changed before the given date.
   * @usage acquia-dam:update-assets --finalized-version
   *   Queue DAM changes + finalized version checks, then process.
   * @usage acquia-dam:update-assets --batch-size=100 --limit=200
   *   Queue in batches of 100 and process up to 200 items.
   * @usage acquia-dam:update-assets --media-type=acquia_dam_image_asset
   *   Update and process only media items of type acquia_dam_image_asset.
   *
   * @help This command provides a complete end-to-end asset update
   *   workflow by combining queue-assets and process-queue commands. Its the
   *   most convenient way to perform a full asset update of Acquia DAM assets.
   *   Use --date-filter to limit assets by date. Supported formats:
   *   "mm/dd/yyyy to mm/dd/yyyy" (between), "after mm/dd/yyyy", or "before mm/dd/yyyy".
   */
  #[CLI\Command(name: 'acquia-dam:update-assets', aliases: ['adua'])]
  #[CLI\Option(name: 'date-filter', description: 'Filter by date. Formats: "mm/dd/yyyy to mm/dd/yyyy" (between), "after mm/dd/yyyy", "before mm/dd/yyyy".')]
  #[CLI\Option(name: 'finalized-version', description: 'Additionally check remaining media for finalized version changes')]
  #[CLI\Option(name: 'batch-size', description: 'Number of media items to process in each batch')]
  #[CLI\Option(name: 'limit', description: 'Maximum number of queue items to process')]
  #[CLI\Option(name: 'media-type', description: 'Specific media type to process')]
  #[CLI\Usage(name: 'acquia-dam:update-assets', description: 'Queue and process all DAM media updates')]
  #[CLI\Usage(name: 'acquia-dam:update-assets --date-filter="06/15/2026 to 07/15/2026"', description: 'Queue and process only assets changed between the given dates')]
  #[CLI\Usage(name: 'acquia-dam:update-assets --date-filter="after 06/15/2026"', description: 'Queue and process only assets changed after the given date')]
  #[CLI\Usage(name: 'acquia-dam:update-assets --date-filter="before 06/15/2026"', description: 'Queue and process only assets changed before the given date')]
  #[CLI\Usage(name: 'acquia-dam:update-assets --finalized-version', description: 'Queue with version checks and process')]
  #[CLI\Usage(name: 'acquia-dam:update-assets --batch-size=100 --limit=200', description: 'Queue in batches of 100 and process up to 200 items')]
  #[CLI\Usage(name: 'acquia-dam:update-assets --media-type=acquia_dam_image_asset', description: 'Update and process only specific media type')]
  #[CLI\Help(description: 'This command provides a complete end-to-end asset
  update workflow by combining the queue-assets and process-queue commands.
  Its the most convenient way to perform a full asset update of Acquia DAM assets.
  Use --date-filter to limit assets by date. Supported formats:
  "mm/dd/yyyy to mm/dd/yyyy" (between), "after mm/dd/yyyy", "before mm/dd/yyyy".')]
  public function updateAssets(
    array $options = [
      'date-filter' => '',
      'finalized-version' => FALSE,
      'batch-size' => 50,
      'limit' => 0,
      'media-type' => NULL,
    ],
  ): void {
    $this->output()->writeln('<info>=== STARTING DAM ASSET UPDATE AND PROCESSING ===</info>');
    $this->output()->writeln('');

    try {
      if (!empty($options['date-filter'])) {
        $this->output()->writeln(sprintf('<comment>Date filter applied: %s</comment>', $options['date-filter']));
      }
      else {
        $this->output()->writeln('<comment>Processing: ALL assets from DAM</comment>');
      }

      if ($options['finalized-version']) {
        $this->output()->writeln('<comment>Additional check: Finalized version verification enabled</comment>');
      }

      $this->output()->writeln('');

      $queue_result = $this->assetUpdater->syncAssets([
        'since-timestamp' => NULL,
        'date-filter' => (string) $options['date-filter'] ?? 'before ' .date('mm/dd/yyyy'),
        'finalized-version' => (bool) $options['finalized-version'],
        'batch-size' => (int) $options['batch-size'],
        'media-type' => $options['media-type'],
      ]);

      // Display queue results.
      $this->output()->writeln('<info>=== DAM API SYNC RESULTS ===</info>');
      $this->output()->writeln(sprintf('Assets reported by DAM: <comment>%d</comment>',
        $queue_result['dam_processed']));
      $this->output()->writeln(sprintf('Media items unpublished: <comment>%d</comment>',
        $queue_result['unpublished']));
      $this->output()->writeln(sprintf('Media items queued from DAM changes: <comment>%d</comment>',
        $queue_result['queued_from_dam']));

      if ($options['finalized-version']) {
        $this->output()->writeln('');
        $this->output()->writeln('<info>=== FINALIZED VERSION CHECK ===</info>');
        $this->output()->writeln(sprintf('Additional media queued from version check: <comment>%d</comment>',
          $queue_result['queued_from_version_check']));
      }

      $total_queued = $queue_result['queued_from_dam'] + $queue_result['queued_from_version_check'];
      $this->output()->writeln('');
      $this->output()->writeln(sprintf('<info>Total queued for processing: %d</info>', $total_queued));

      if ($total_queued === 0) {
        $this->output()->writeln('<info>No updates needed - all media items are up to date.</info>');
        $this->output()->writeln('');
        $this->output()->writeln('<info>=== DAM ASSET UPDATE COMPLETED ===</info>');
        return;
      }

      // Step 2: Process the queue.
      $this->output()->writeln('');
      $this->output()->writeln('<info>=== PROCESSING UPDATE QUEUE ===</info>');

      $process_result = $this->assetUpdater->processQueue($options);

      // Display process results.
      $this->output()->writeln('');
      $this->output()->writeln('<info>=== QUEUE PROCESSING SUMMARY ===</info>');
      $this->output()->writeln(sprintf('Items processed: <comment>%d</comment>', $process_result['processed']));
      $this->output()->writeln(sprintf('Remaining items in queue: <comment>%d</comment>', $process_result['remaining']));

      if ($process_result['errors'] > 0) {
        $this->output()->writeln(sprintf('Errors encountered: <error>%d</error>', $process_result['errors']));
      }

    }
    catch (\Exception $e) {
      $this->output()->writeln(sprintf('<error>An error occurred: %s</error>', $e->getMessage()));
      $this->logger()->error('Error in DAM update command: @message', ['@message' => $e->getMessage()]);
    }

    $this->output()->writeln('');
    $this->output()->writeln('<info>=== DAM ASSET UPDATE AND PROCESSING COMPLETED ===</info>');
  }

  /**
   * Resolve and update media to correct media type.
   *
   * This command iterates through all media items that use Acquia DAM assets
   * and checks if they are assigned to the correct media type based on the
   * asset metadata in DAM. If a media item is found to be assigned to an
   * incorrect media type, it updates the media item to the correct type.
   *
   * @param array $options
   *   An associative array of options whose values come from cli, aliases,
   *   config, etc.
   *
   * @command acquia-dam:resolve-asset-media-type
   * @aliases adramt
   *
   * @option batch-size
   *   Number of media items to process in each batch max 200. Defaults to 50.
   * @option media-type
   *   Specific media type to process.
   *   If not specified, all DAM media types will be processed.
   * @usage acquia-dam:resolve-asset-media-type --batch-size=100
   *   Process in batches of 100 items.
   * @usage acquia-dam:resolve-asset-media-type --media-type=acquia_dam_image_asset
   *   Process only media items of type acquia_dam_image_asset.
   * @help This command resolves and updates media items to their correct
   *   media types based on the asset metadata in Acquia DAM. It processes
   *   media items in batches and provides a summary of the number of processed
   *   and updated media items.
   */
  #[CLI\Command(name: 'acquia-dam:resolve-asset-media-type', aliases: ['adramt'])]
  #[CLI\Option(name: 'batch-size', description: 'Number of media items to process in each batch max 200.')]
  #[CLI\Option(name: 'media-type', description: 'Specific media type to process. If not specified, all DAM media types will be processed.')]
  #[CLI\Usage(name: 'acquia-dam:resolve-asset-media-type', description: 'Resolve and update media to correct media types.')]
  #[CLI\Usage(name: 'acquia-dam:resolve-asset-media-type --batch-size=100', description: 'Process in batches of 100 items.')]
  #[CLI\Usage(name: 'acquia-dam:resolve-asset-media-type --media-type=acquia_dam_image_asset', description: 'Process only media items of type acquia_dam_image_asset.')]
  #[CLI\Help(description: 'This command resolves and updates media items to
  their correct media types based on the asset metadata in Acquia DAM.
  It processes media items in batches and provides a summary of the number
  of processed and updated media items.')]
  public function resolveAssetMediaType(
    array $options = [
      'batch-size' => 50,
      'media-type' => NULL,
    ],
  ): void {
    $media_storage = $this->entityTypeManager->getStorage('media');
    $query = $media_storage->getQuery()
      ->accessCheck(FALSE)
      ->condition('acquia_dam_asset_id', NULL, 'IS NOT NULL');

    if ($options['media-type']) {
      $query->condition('bundle', $options['media-type']);
    }

    $media_ids = $query->execute();

    if (empty($media_ids)) {
      $this->output()->writeln(sprintf('No media items found to process.'));
      return;
    }

    // Validate and set batch size with upper limit.
    $batch_size = (int) $options['batch-size'];
    // Cap batch size between 1 and 200.
    $batch_size = min($batch_size <= 0 ? 50 : $batch_size, 200);

    $this->output()->writeln(sprintf('Processing %d media items in batches of %d',
      count($media_ids),
      $batch_size
    ));

    // Initialize the DAM client once outside the loop.
    try {
      $dam_client = $this->acquiaDamClientFactory->getSiteClient();
    }
    catch (\Exception $e) {
      $this->output()->writeln(sprintf('<error>Failed to initialize DAM client: %s</error>', $e->getMessage()));
      return;
    }

    $chunks = array_chunk($media_ids, $batch_size);
    $processed = 0;
    $updated = 0;
    $errors = 0;
    $total_chunks = count($chunks);

    // Pre-load all media type configurations to avoid repeated queries.
    $media_type_storage = $this->entityTypeManager->getStorage('media_type');
    $all_media_types = $media_type_storage->loadMultiple();
    $media_type_configs = [];
    foreach ($all_media_types as $type_id => $type) {
      $media_type_configs[$type_id] = [
        'source' => $type->get('source'),
        'download_assets' => $type->get('source_configuration')['download_assets'] ?? FALSE,
      ];
    }

    foreach ($chunks as $chunk_index => $chunk) {
      $this->output()->writeln(sprintf('Processing chunk %d/%d (%d items)',
        $chunk_index + 1,
        $total_chunks,
        count($chunk)
      ));

      try {
        // Fetch old table data for this chunk only.
        $old_data_by_entity_image = $this->fetchOldTableData(self::MEDIA_IMAGE_TABLE, $chunk);
        $old_data_by_entity_file = $this->fetchOldTableData(self::MEDIA_FILE_TABLE, $chunk);

        // Use loadMultiple with valid media IDs only.
        /** @var \Drupal\media\MediaInterface[] $media_items */
        $media_items = $media_storage->loadMultiple($chunk);

        foreach ($media_items as $media_id => $media) {
          try {
            $result = $this->processMediaItem(
              $media,
              $media_id,
              $media_type_configs,
              $old_data_by_entity_image,
              $old_data_by_entity_file,
              $media_storage,
              $dam_client
            );

            if ($result['updated']) {
              $updated++;
            }
            $processed++;
          }
          catch (\Exception $e) {
            $this->output()->writeln(sprintf('Error processing media %d: %s',
              $media_id,
              $e->getMessage()
            ));
            $this->logger()->error('Error processing media @id: @message', [
              '@id' => $media_id,
              '@message' => $e->getMessage(),
            ]);
            $errors++;
            $processed++;
          }
        }

        // Clear entity cache for processed items to free memory.
        $media_storage->resetCache($chunk);

        // Force garbage collection every 10 chunks to manage memory.
        if (($chunk_index + 1) % 10 === 0) {
          gc_collect_cycles();
        }

      }
      catch (\Exception $e) {
        $this->output()->writeln(sprintf('Error processing chunk %d: %s',
          $chunk_index + 1,
          $e->getMessage()
        ));
        $this->logger()->error('Error processing chunk @chunk: @message', [
          '@chunk' => $chunk_index + 1,
          '@message' => $e->getMessage(),
        ]);
        $errors += count($chunk);
        $processed += count($chunk);
        // Skip this chunk but continue with the next one.
        continue;
      }
    }

    // Final summary.
    $this->output()->writeln('');
    $this->output()->writeln('<info>=== MEDIA TYPE RESOLUTION SUMMARY ===</info>');
    $this->output()->writeln(sprintf('Media items processed: <comment>%d</comment>', $processed));
    $this->output()->writeln(sprintf('Media items updated: <comment>%d</comment>', $updated));

    if ($errors > 0) {
      $this->output()->writeln(sprintf('Errors encountered: <error>%d</error>', $errors));
    }
  }

  /**
   * Fetch old table data for a chunk of media IDs.
   */
  private function fetchOldTableData(string $table_name, array $media_ids): array {
    $old_data_by_entity = [];

    $old_table_data = $this->connection->select($table_name, 'odt')
      ->fields('odt')
      ->condition('entity_id', $media_ids, 'IN')
      ->execute()
      ->fetchAll();

    foreach ($old_table_data as $row) {
      $old_data_by_entity[$row->entity_id][] = $row;
    }

    return $old_data_by_entity;
  }

  /**
   * Process a single media item for media type resolution.
   *
   * @param \Drupal\media\MediaInterface $media
   *   The media entity to process.
   * @param int $media_id
   *   The media entity ID.
   * @param array $media_type_configs
   *   Pre-loaded media type configurations.
   * @param array $old_data_by_entity_image
   *   Old data from the image field table indexed by entity ID.
   * @param array $old_data_by_entity_file
   *   Old data from the file field table indexed by entity ID.
   * @param \Drupal\Core\Entity\EntityStorageInterface $media_storage
   *   The media entity storage.
   * @param \Drupal\acquia_dam\Client\AcquiaDamClient $dam_client
   *   The Acquia DAM client.
   */
  private function processMediaItem(
    MediaInterface $media,
    int $media_id,
    array $media_type_configs,
    array $old_data_by_entity_image,
    array $old_data_by_entity_file,
    EntityStorageInterface $media_storage,
    AcquiaDamClient $dam_client,
  ): array {
    if (!$media->hasField('acquia_dam_asset_id')) {
      return ['updated' => FALSE];
    }

    $old_media_type_id = $media->bundle();
    $asset_id_field = $media->get('acquia_dam_asset_id')->getValue();
    if (empty($asset_id_field[0]['asset_id'])) {
      return ['updated' => FALSE];
    }

    $asset_id = $asset_id_field[0]['asset_id'];

    try {
      $asset_data = $dam_client->getAsset($asset_id);
      $new_media_type = $this->mediaTypeResolver->resolve($asset_data);
    }
    catch (\Exception $e) {
      $this->output()->writeln(sprintf('Failed to resolve media type for media %d: %s',
        $media_id,
        $e->getMessage()
      ));
      return ['updated' => FALSE];
    }

    if (!$new_media_type) {
      $this->output()->writeln(sprintf('Could not determine new media type for media %d.', $media_id));
      return ['updated' => FALSE];
    }

    $new_media_type_id = $new_media_type->id();

    // Skip if media type is already correct.
    if ($old_media_type_id === $new_media_type_id) {
      return ['updated' => FALSE];
    }

    // Use pre-loaded configurations.
    $old_config = $media_type_configs[$old_media_type_id] ?? NULL;
    $new_config = $media_type_configs[$new_media_type_id] ?? NULL;

    if (!$old_config || !$new_config) {
      $this->output()->writeln(sprintf('Missing media type configuration for media %d', $media_id));
      return ['updated' => FALSE];
    }

    // Create a new revision to preserve bundle change history.
    $media->setNewRevision(TRUE);
    $media->revision_log = sprintf(
      'Bundle changed from %s to %s via DAM asset type resolution.',
      $old_media_type_id,
      $new_media_type_id
    );

    // Update the media bundle.
    $media->set('bundle', $new_media_type_id);
    $media->save();

    $this->output()->writeln(sprintf('Updated media %d from %s to %s',
      $media_id,
      $old_media_type_id,
      $new_media_type_id
    ));

    // Get the new revision ID after save.
    $new_revision_id = $media->getRevisionId();

    // Clear media cache.
    $media_storage->resetCache([$media_id]);

    // Update bundle column in media_field_data table.
    $this->connection->update('media_field_data')
      ->fields(['bundle' => $new_media_type_id])
      ->condition('mid', $media_id)
      ->execute();

    // Update bundle column in media__acquia_dam_asset_id table.
    $this->connection->update('media__acquia_dam_asset_id')
      ->fields(['bundle' => $new_media_type_id])
      ->condition('entity_id', $media_id)
      ->execute();

    // Update bundle column in media_revision__acquia_dam_asset_id table.
    $this->connection->update('media_revision__acquia_dam_asset_id')
      ->fields(['bundle' => $new_media_type_id])
      ->condition('entity_id', $media_id)
      ->condition('revision_id', $new_revision_id)
      ->execute();

    // Handle file migration if needed.
    $this->handleFileDataMigration(
      $media_id,
      $media,
      $old_config,
      $new_config,
      $old_data_by_entity_image,
      $old_data_by_entity_file,
      $new_media_type_id,
      $asset_data,
      $asset_id
    );

    return ['updated' => TRUE];
  }

  /**
   * Handle file data migration between media types.
   */
  private function handleFileDataMigration(
    int $media_id,
    $media,
    array $old_config,
    array $new_config,
    array $old_data_by_entity_image,
    array $old_data_by_entity_file,
    string $new_media_type_id,
    $asset_data,
    string $asset_id,
  ): void {
    $old_render_method = $old_config['download_assets'];
    $new_render_method = $new_config['download_assets'];
    $source = $new_config['source'];

    // Migrate local file data between tables when the old media type
    // had "download and sync" enabled, preserving downloaded assets.
    if ($old_render_method) {
      if (isset($old_data_by_entity_image[$media_id]) || isset($old_data_by_entity_file[$media_id])) {
        // Update bundle in old tables for media items
        // that are switching media type other than image.
        if ($source != 'acquia_dam_asset:image' && isset($old_data_by_entity_file[$media_id])) {
          $this->connection->update(self::MEDIA_FILE_TABLE)
            ->fields(['bundle' => $new_media_type_id])
            ->condition('entity_id', $media_id)
            ->execute();
        }
        // Media type transitions between image and file type by migrating data.
        else {
          $this->migrateFileDataBetweenTables(
            $media_id,
            $source,
            $new_media_type_id,
            $old_data_by_entity_image,
            $old_data_by_entity_file
          );
        }
      }
    }
    else {
      // If the old media type did not have "download and sync" enabled,
      // but new media type does, we need to download the asset now.
      if ($new_render_method) {
        $this->assetFileHelper->downloadFile($media, $media->getSource(), $asset_data, $asset_id);
      }
    }
  }

  /**
   * Migrate file data between image and file tables.
   */
  private function migrateFileDataBetweenTables(
    int $media_id,
    string $source,
    string $new_media_type_id,
    array $old_data_by_entity_image,
    array $old_data_by_entity_file,
  ): void {
    $new_table = $source === 'acquia_dam_asset:image' ? self::MEDIA_IMAGE_TABLE : self::MEDIA_FILE_TABLE;
    $old_table = $source === 'acquia_dam_asset:image' ? self::MEDIA_FILE_TABLE : self::MEDIA_IMAGE_TABLE;
    $move_array = $source === 'acquia_dam_asset:image' ? $old_data_by_entity_file : $old_data_by_entity_image;
    $new_target_field = $source === 'acquia_dam_asset:image' ? 'acquia_dam_managed_image_target_id' : 'acquia_dam_managed_file_target_id';
    $old_target_field = $source === 'acquia_dam_asset:image' ? 'acquia_dam_managed_file_target_id' : 'acquia_dam_managed_image_target_id';

    // Insert all rows for this entity in one batch.
    if (isset($move_array[$media_id])) {
      foreach ($move_array[$media_id] as $result) {
        $this->connection->insert($new_table)
          ->fields([
            'bundle' => $new_media_type_id,
            'entity_id' => $result->entity_id,
            'revision_id' => $result->revision_id,
            'langcode' => $result->langcode,
            'delta' => $result->delta,
            $new_target_field => $result->$old_target_field,
          ])->execute();
      }

      // Delete from old table.
      $this->connection->delete($old_table)
        ->condition('entity_id', $media_id)
        ->execute();
    }
  }

  /**
   * Parse the --date-filter option value into after/before timestamps.
   *
   * @param string $date_filter
   *   The raw date filter string from the CLI option.
   *
   * @return array
   *   An associative array with optional keys 'after' and 'before' as
   *   Unix timestamps.
   *
   * @throws \InvalidArgumentException
   *   When the date filter format is not recognized.
   */
  private function parseDateFilter(string $date_filter): array {
    $result = ['after' => NULL, 'before' => NULL];

    // Between: "mm/dd/yyyy to mm/dd/yyyy".
    if (preg_match('/^(\d{2}\/\d{2}\/\d{4})\s+to\s+(\d{2}\/\d{2}\/\d{4})$/', $date_filter, $matches)) {
      $result['after'] = strtotime($matches[1]);
      $result['before'] = strtotime($matches[2]);
    }
    // After: "after mm/dd/yyyy".
    elseif (preg_match('/^after\s+(\d{2}\/\d{2}\/\d{4})$/', $date_filter, $matches)) {
      $result['after'] = strtotime($matches[1]);
    }
    // Before: "before mm/dd/yyyy".
    elseif (preg_match('/^before\s+(\d{2}\/\d{2}\/\d{4})$/', $date_filter, $matches)) {
      $result['before'] = strtotime($matches[1]);
    }
    else {
      throw new \InvalidArgumentException(sprintf(
        'Invalid --date-filter format: "%s". Valid formats: "mm/dd/yyyy to mm/dd/yyyy", "after mm/dd/yyyy", "before mm/dd/yyyy".',
        $date_filter
      ));
    }

    return $result;
  }

}
