data_pipelines-1.x-dev/src/Form/DatasetBatchOperations.php

<?php

declare(strict_types=1);

namespace Drupal\data_pipelines\Form;

use Drupal\Component\Render\MarkupInterface;
use Drupal\Core\Batch\BatchBuilder;
use Drupal\Core\DependencyInjection\ContainerInjectionInterface;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Queue\QueueFactory;
use Drupal\Core\Queue\QueueWorkerManagerInterface;
use Drupal\Core\Queue\RequeueException;
use Drupal\Core\StringTranslation\TranslatableMarkup;
use Drupal\data_pipelines\Entity\Dataset;
use Drupal\data_pipelines\Entity\DatasetInterface;
use Symfony\Component\DependencyInjection\ContainerInterface;

/**
 * Defines a class for building and running dataset batched operations.
 */
class DatasetBatchOperations implements ContainerInjectionInterface {

  /**
   * Constructs a new PreviewSiteBuilder.
   *
   * @param \Drupal\Core\Queue\QueueFactory $queueFactory
   *   Queue factory.
   * @param \Drupal\Core\Queue\QueueWorkerManagerInterface $queueManager
   *   Queue manager.
   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entityTypeManager
   *   Entity type manager.
   */
  final public function __construct(protected QueueFactory $queueFactory, protected QueueWorkerManagerInterface $queueManager, protected EntityTypeManagerInterface $entityTypeManager) {
  }

  /**
   * {@inheritdoc}
   */
  public static function create(ContainerInterface $container) {
    return new static(
      $container->get('queue'),
      $container->get('plugin.manager.queue_worker'),
      $container->get('entity_type.manager'),
    );
  }

  /**
   * Factory method.
   *
   * @return \Drupal\data_pipelines\Form\DatasetBatchOperations
   *   New instance.
   */
  protected static function factory(): DatasetBatchOperations {
    $instance = \Drupal::classResolver(self::class);
    assert($instance instanceof DatasetBatchOperations);
    return $instance;
  }

  /**
   * Gets batch for dataset.
   *
   * @param \Drupal\data_pipelines\Entity\DatasetInterface $dataset
   *   Dataset.
   * @param bool $saving
   *   TRUE if Dataset is being saved FALSE otherwise.
   *
   * @return array
   *   Batch.
   */
  public static function batchForDataset(DatasetInterface $dataset, bool $saving = FALSE): array {
    if (!$dataset->isPublished()) {
      return [];
    }
    $batch_builder = new BatchBuilder();
    $batch_builder
      ->setTitle(new TranslatableMarkup('Validating and indexing dataset %name', [
        '%name' => $dataset->label(),
      ]))
      ->setInitMessage(new TranslatableMarkup('Preparing...'))
      ->setProgressive(TRUE)
      ->setFinishCallback([self::class, 'finished'])
      ->setProgressMessage(new TranslatableMarkup('Step @current of @total'))
      ->addOperation([self::class, 'operationQueueItem'], [$dataset->id()])
      ->addOperation([self::class, 'operationProcess'], [$dataset->id()]);
    return $batch_builder->toArray();
  }

  /**
   * Batch callback.
   */
  public static function operationQueueItem(int $dataset_id, &$context): void {
    $context['results']['dataset_id'] = $dataset_id;
    $dataset = Dataset::load($dataset_id);
    assert($dataset instanceof DatasetInterface);
    if (!$context['sandbox']) {
      $context['sandbox']['context'] = new DatasetBatchContext($dataset->getBatchSize());
      $dataset->resetLogs();
      $dataset->setPendingProcessing();
    }
    $inner_context = &$context['sandbox']['context'];
    $context['finished'] = 0;
    $dataset->queueProcessing(\Drupal::queue($dataset->getProcessingQueueId()), $inner_context);
    $context['message'] = new TranslatableMarkup('Queueing processing, queued @count records.', ['@count' => $inner_context->getProcessed()]);
    if ($inner_context->isFinished()) {
      $context['finished'] = 1;
      $context['message'] = new TranslatableMarkup('Data is successfully queued for processing.');
    }
  }

  /**
   * Batch callback.
   */
  public static function operationProcess(int $dataset_id, &$context): void {
    $dataset = Dataset::load($dataset_id);
    assert($dataset instanceof DatasetInterface);
    if ($dataset->isPendingProcessing()) {
      $remaining = self::factory()->process($dataset);
      self::updateFinishedPercent($remaining, $context);
      $total = $context['sandbox']['total'];
      $context['message'] = new TranslatableMarkup('Processed @count of @total (@percent %) chunks, @remaining remaining.', [
        '@count' => number_format($total - $remaining),
        '@total' => number_format($total),
        '@percent' => $total > 0 ? round(100 * ($total - $remaining) / $total) : 0,
        '@remaining' => number_format($remaining),
      ]);
      if ($remaining === 0) {
        $dataset->setProcessingComplete();
        $context['message'] = new TranslatableMarkup('Processed dataset.');
        return;
      }
      return;
    }
    $context['finished'] = 1;
    $context['message'] = new TranslatableMarkup('Skipped processing dataset, validation errors exist.');
  }

  /**
   * Index a dataset.
   *
   * @param \Drupal\data_pipelines\Entity\DatasetInterface $dataset
   *   Dataset.
   *
   * @return int
   *   Number remaining.
   */
  protected function process(DatasetInterface $dataset): int {
    return $this->processQueueItem($dataset->getProcessingQueueId());
  }

  /**
   * Processes a queue item.
   *
   * @param string $queue_name
   *   Queue name.
   *
   * @return int
   *   Number of remaining items.
   */
  protected function processQueueItem(string $queue_name) : int {
    $queue = $this->queueFactory->get($queue_name);
    $item = $queue->claimItem();
    if (!$item) {
      return 0;
    }
    $queue->createQueue();
    $worker = $this->queueManager->createInstance($queue_name);
    try {
      $worker->processItem($item->data);
      $queue->deleteItem($item);
    }
    catch (RequeueException $e) {
      $queue->releaseItem($item);
    }
    return $queue->numberOfItems();
  }

  /**
   * Batch finished callback.
   */
  public static function finished(bool $success, array $results) {
    if (!empty($results['dataset_id'])) {
      $dataset = Dataset::load($results['dataset_id']);
      assert($dataset instanceof DatasetInterface);
      $log_count = count($dataset->getLogs());
      if ($dataset->isProcessed() && $log_count === 0) {
        $message = new TranslatableMarkup('Dataset %name was successfully processed.', ['%name' => $dataset->label()]);
        static::log($message);
        return;
      }
      $message = new TranslatableMarkup('Dataset %name was processed but some rows failed validation or skipped and were not processed. Please check <a href=":url">the logs</a>.', [
        '%name' => $dataset->label(),
        ':url' => $dataset->toUrl()->toString(),
      ]);
      static::log($message, TRUE);
      return;
    }
    $message = new TranslatableMarkup('The validation and processing operation did not complete.');
    static::log($message, TRUE);
  }

  /**
   * A method to facilitate logging.
   *
   * @param \Drupal\Component\Render\MarkupInterface $message
   *   The message.
   * @param bool $error
   *   A boolean indicating whether the log is an error.
   */
  protected static function log(MarkupInterface $message, bool $error = FALSE): void {
    \Drupal::logger('data_pipelines')->{$error ? 'warning' : 'info'}($message);
    \Drupal::messenger()->{$error ? 'addError' : 'addMessage'}($message);
  }

  /**
   * Updates finished percent.
   *
   * @param int $remaining
   *   Remaining items.
   * @param array $context
   *   Batch context.
   */
  protected static function updateFinishedPercent(int $remaining, &$context): void {
    if ($remaining === 0) {
      $context['finished'] = 1;
      $context['sandbox']['total'] = 1;
      return;
    }
    if (!isset($context['sandbox']['total'])) {
      $context['sandbox']['total'] = $remaining + 1;
    }
    $context['finished'] = ($context['sandbox']['total'] - $remaining) / $context['sandbox']['total'];
  }

}