<?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')
    );
  }

  /**
   * Update all DAM media items by checking their assets for changes.
   *
   * This command iterates through all media items that use Acquia DAM assets
   * and checks if they need to be updated based on changes in the 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 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 --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.
   * @usage acquia-dam:queue-update-assets --finalized-version
   *   Queue only media items if the finalized version differs.
   * @help This command checks all DAM media items for updates based on changes
   *   in Acquia DAM. It processes media items in batches and provides a summary
   *   of the number of media items verified for update and those queued for
   *   update.
   */
  #[CLI\Command(name: 'acquia-dam:queue-update-assets', aliases: ['adqua'])]
  #[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\Option(name: 'finalized-version', description: 'Check and queue media for updates if the finalized version differs.')]
  #[CLI\Usage(name: 'acquia-dam:queue-update-assets', description: 'Update all DAM media items')]
  #[CLI\Usage(name: 'acquia-dam:queue-update-assets --batch-size=100', description: 'Update all DAM media items in batches of 100')]
  #[CLI\Usage(name: 'acquia-dam:queue-update-assets --media-type=acquia_dam_image_asset', description: 'Queue only media items of type acquia_dam_image_asset')]
  #[CLI\Usage(name: 'acquia-dam:queue-update-assets --finalized-version', description: 'Queue only media items if the finalized version differs.')]
  #[CLI\Help(description: 'This command checks all DAM media items for updates
  based on changes in Acquia DAM. It processes media items in batches and provides
  a summary of the number of media items verified for update and those queued
  for update.')]
  public function queueAssets(
    array $options = [
      'batch-size' => 50,
      'media-type' => NULL,
      'finalized-version' => FALSE,
    ],
  ): void {
    try {
      $result = $this->assetUpdater->queueAssets($options);

      if (empty($result['media_types'])) {
        $this->output()->writeln('<error>No DAM media types found.</error>');
        return;
      }

      $this->output()->writeln(sprintf('<info>Found %d DAM media type(s): %s</info>',
        count($result['media_types']),
        implode(', ', $result['media_types'])
      ));

      // Summary.
      $this->output()->writeln('');
      $this->output()->writeln('<info>=== UPDATE SUMMARY ===</info>');
      $this->output()->writeln(sprintf('Media items verified for update: <comment>%d</comment>', $result['total_processed']));
      $this->output()->writeln(sprintf('Media items queued for update: <comment>%d</comment>', $result['total_updated']));

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

      if ($result['total_updated'] > 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>');
      }

    }
    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()]);
    }
  }

  /**
   * 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 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 --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.
   * @usage acquia-dam:update-assets --finalized-version
   *   Update and process only media items if the finalized version differs.
   * @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.
   */
  #[CLI\Command(name: 'acquia-dam:update-assets', aliases: ['adua'])]
  #[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\Option(name: 'finalized-version', description: 'Check and queue media for updates if the finalized version differs.')]
  #[CLI\Usage(name: 'acquia-dam:update-assets', description: 'Queue and process all DAM media updates')]
  #[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 media items of type acquia_dam_image_asset')]
  #[CLI\Usage(name: 'acquia-dam:update-assets --finalized-version', description: 'Update and process only media items if the finalized version differs')]
  #[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.')]
  public function updateAssets(
    array $options = [
      'batch-size' => 50,
      'limit' => 0,
      'media-type' => NULL,
      'finalized-version' => FALSE,
    ],
  ): void {
    $this->output()->writeln('<info>=== STARTING DAM ASSET UPDATE AND PROCESSING ===</info>');
    $this->output()->writeln('');

    // Set up progress callback for detailed output.
    $progress_callback = function ($progress_info) {
      $type_map = [
        'info' => 'info',
        'success' => 'info',
        'comment' => 'comment',
        'error' => 'error',
        'progress' => 'comment',
      ];

      $tag = $type_map[$progress_info['type']] ?? 'info';
      $this->output()->writeln(sprintf('<%s>%s</%s>', $tag, $progress_info['message'], $tag));
    };

    try {
      // Queue assets.
      $queue_result = $this->assetUpdater->queueAssets($options);

      // Display queue results.
      if (!empty($queue_result['media_types'])) {
        $this->output()->writeln(sprintf('<info>Found %d DAM media type(s): %s</info>',
          count($queue_result['media_types']),
          implode(', ', $queue_result['media_types'])
        ));

        // Process each media type with progress reporting.
        foreach ($queue_result['media_types'] as $media_type) {
          $this->output()->writeln(sprintf('<info>Processing media type: %s</info>', $media_type));
          $this->assetUpdater->processMediaType(
            $media_type,
            (int) $options['batch-size'],
            $options['finalized-version'],
            $progress_callback
          );
        }
      }

      // Process the queue.
      $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];
    }

    // 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
    ));

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

    // Update bundle column in field table.
    $this->connection->update('media__acquia_dam_asset_id')
      ->fields(['bundle' => $new_media_type_id])
      ->condition('entity_id', $media_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();
    }
  }

}
