lionbridge_translation_provider-8.x-2.4/tmgmt_contentapi/src/Controller/QueueProcessController.php

<?php

namespace Drupal\tmgmt_contentapi\Controller;

use Drupal\Core\Controller\ControllerBase;
use Symfony\Component\HttpFoundation\JsonResponse;
use Symfony\Component\DependencyInjection\ContainerInterface;
use Drupal\Core\Access\CsrfTokenGenerator;

/**
 * @file
 * Contains \Drupal\tmgmt_contentapi\Controller\QueueProcessController.
 *
 * This controller provides an endpoint for processing specific Drupal queues
 * in batches, typically initiated programmatically or via API requests.
 * It validates CSRF tokens, queue names, and batch sizes, then processes
 * items from the queue using the appropriate queue worker plugin. Logging
 * is performed for all major actions and errors. Intended for use in the
 * Lionbridge Translation Provider integration.
 */
/**
 * Controller for processing the queue.
 *
 * This will help to process the queue, if initiated programmatically.
 */
class QueueProcessController extends ControllerBase {

  /**
   * The CSRF token service.
   *
   * @var \Drupal\Core\Access\CsrfTokenGenerator
   */
  protected $csrfToken;
  /**
   * The request stack service.
   *
   * @var \Symfony\Component\HttpFoundation\RequestStack
   */
  protected $requestStack;
  /**
   * The queue manager service.
   *
   * @var \Drupal\Core\Queue\QueueFactoryInterface
   */
  protected $queueManager;
  /**
   * The queue worker manager service.
   *
   * @var \Drupal\Core\Queue\QueueWorkerManagerInterface
   */
  protected $queueWorkerManager;

  public function __construct($csrfToken, $requestStack, $queueManager, $queueWorkerManager) {
    $this->csrfToken = $csrfToken;
    $this->requestStack = $requestStack;
    $this->queueManager = $queueManager;
    $this->queueWorkerManager = $queueWorkerManager;
  }

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

  /**
   * Process the queue in batches.
   */
  public function processQueueInBackground($queue_name, int $batch_size = 100) {
    $request = $this->requestStack->getCurrentRequest();
    $token = $request->headers->get('X-CSRF-Token');
    if (empty($token)) {
      return new JsonResponse(['error' => 'CSRF token is missing'], 403);
    }
    if (!$this->csrfToken->validate($token, 'queue_process_background')) {
      return new JsonResponse(['error' => 'Invalid CSRF token'], 403);
    }

    if (!preg_match('/^[a-zA-Z0-9_]+$/', $queue_name)) {
      return new JsonResponse(['error' => 'Invalid queue name.'], 400);
    }

    if ($batch_size <= 0 || $batch_size > 1000) {
      return new JsonResponse(['error' => 'Batch size must be a positive integer and not exceed 1000.'], 400);
    }

    $queue_array = [
      \Drupal::service('tmgmt_contentapi.create_job')::QUEUE_NAME_EXPORT_JOBS,
      \Drupal::service('tmgmt_contentapi.create_job')::QUEUE_NAME_SEND_FILES,
      \Drupal::service('tmgmt_contentapi.create_job')::QUEUE_NAME_GENERATE_FILES,
      \Drupal::service('tmgmt_contentapi.import_job')::QUEUE_WORKER_IMPORT_TRANSLATION_JOBS_TO_CAPI,
      \Drupal::service('tmgmt_contentapi.import_job')::QUEUE_NAME_IMPORT_JOBS,
      \Drupal::service('tmgmt_contentapi.import_job')::QUEUE_NAME_IMPORT_JOBS_MANUALLY,
      \Drupal::service('tmgmt_contentapi.capi_data_processor')::QUEUE_NAME_MIGRATE_JOBS
    ];

    if (!in_array($queue_name, $queue_array)) {
      \Drupal::logger('TMGMT_CONTENTAPI')->error('Invalid queue: @queue', ['@queue' => $queue_name]);
      return new JsonResponse(['error' => 'Queue is not valid for this operation.'], 400);
    }

    $queue = $this->queueManager->get($queue_name);
    try {
      $queue_worker = $this->queueWorkerManager->createInstance($queue_name);
    }
    catch (\Exception $e) {
      \Drupal::logger('TMGMT_CONTENTAPI')->error('Queue worker not found: @message', ['@message' => $e->getMessage()]);
      return new JsonResponse(['error' => 'Queue worker not found.'], 500);
    }

    $processed = 0;
    \Drupal::logger('TMGMT_CONTENTAPI')->notice('Queue processing started in background: @queue', ['@queue' => $queue_name]);

    if (!$queue->numberOfItems()) {
      \Drupal::logger('TMGMT_CONTENTAPI')->notice('No Items to process in queue: @queue', ['@queue' => $queue_name]);
      return new JsonResponse(['message' => 'No item found to process']);
    }

    while ($processed < $batch_size && ($item = $queue->claimItem())) {
      try {
        $queue_worker->processItem($item->data);
        $queue->deleteItem($item);
      }
      catch (\Exception $e) {
        \Drupal::logger('TMGMT_CONTENTAPI')->error('Error processing task: @message', ['@message' => $e->getMessage()]);
        $queue->releaseItem($item);
      }
      unset($item);
      gc_collect_cycles();
      $processed++;
    }

    unset($queue_worker);
    unset($queue);
    return new JsonResponse(['message' => 'Processed ' . $processed . ' items.']);
  }

}