ltsecurityinc/app/code/JR/NetSuite/Queue/AbstractQueueManager.php

<?php

namespace JR\NetSuite\Queue;

use JR\NetSuite\Lib\NetSuiteAPI;
use Magento\Framework\Serialize\SerializerInterface;

abstract class AbstractQueueManager {

  /**
   * @var string
   */
  protected $name = 'Abstract';

  /**
   * @var \JR\NetSuite\Helper\Data
   */
  protected $helper;

  /**
   * @var \JR\NetSuite\Logger\Logger
   */
  protected $logger;

  /**
   * @var SerializerInterface
   */
  protected $serializer;

  /**
   * @var NetSuiteAPI
   */
  protected $netSuiteAPI;

  /**
   * AbstractQueueManager constructor.
   *
   * @param \JR\NetSuite\Helper\Data $helper
   * @param \JR\NetSuite\Logger\Logger $logger
   * @param SerializerInterface $serializer
   * @param NetSuiteAPI $netSuiteAPI
   * @return void
   */
  public function __construct(
    \JR\NetSuite\Helper\Data $helper,
    \JR\NetSuite\Logger\Logger $logger,
    SerializerInterface $serializer,
    NetSuiteAPI $netSuiteAPI
  ) {
    $this->helper = $helper;
    $this->logger = $logger;
    $this->serializer = $serializer;
    $this->netSuiteAPI = $netSuiteAPI;
  }

  /**
   * Fetch data from NetSuite and add it to the queue.
   *
   * @return array
   */
  public function fetchAndEnqueue($fetchAll = false) {
    $this->logger->info("[$this->name" . "QueueManager] fetchAndEnqueue(" . ($fetchAll ? "true" : "false") . ") started.");
    // max page to fetch and loop to fetch all pages
    $maxPage = 1;
    $page = 1;
    $pageSize = $this->getPageSize();
    $importSize = 0;

    // Fetch data from NetSuite
    while ($maxPage >= $page) {
      $result = $this->fetch($pageSize, $page);
      if (isset($result['error'])) {
        $this->logger->error($result['message']);
        return [
          'error' => true,
          'message' => $result['message']
        ];
      }
      $data = $this->convertToMagentoFormat($result['data']);

      try {
        $this->enqueue($data);
      } catch (\Exception $e) {
        $this->logger->error("[$this->name" . "QueueManager] " . $e->getMessage());
        throw $e;
      }

      $importSize += count($data);
      if ($fetchAll) {
        $maxPage = $result['totalPages'];
      }
      $page++;
      sleep(30);
    }
    $this->logger->info("[$this->name" . "QueueManager] fetchAndEnqueue(" . ($fetchAll ? "true" : "false") . ") completes with " . $maxPage . " page(s), pageSize = " . $pageSize . ".  " . $this->name);

    return [
      'error' => false,
      'message' => 'Imported ' . $importSize . ' ' . $this->name . ' records.'
    ];
  }

  /**
   * Fetch data from NetSuite
   *
   * @param int $pageSize
   * @param int $page
   * @return array
   */
  abstract public function fetch($pageSize = 50, $page = 1);

  /**
   * get the page size for each fetch.
   *
   * @return int
   */
  abstract public function getPageSize();

  /**
   * Convert NetSuite data to Magento format
   *
   * @param array $data
   * @return array
   */
  abstract public function convertToMagentoFormat($data);

  /**
   * adds data to the queue.
   *
   * @param array $data
   * @return void
   */
  abstract public function enqueue($data);

  /**
   * Process batch of queue items
   *
   * @param int $pageSize
   * @param int $page
   * @return
   */
  public function processBatch($batchSize = 50) {
    $this->logger->info("[$this->name" . "QueueManager] processBatch: batch size -> " . $batchSize);
    $queueItems = $this->dequeue($batchSize);
    $this->logger->info("[$this->name" . "QueueManager] processBatch: queue item size -> " . count($queueItems));
    $totalProcessed = 0;
    if (count($queueItems) == 0) {
      return $totalProcessed;
    }

    $memoryStart = memory_get_usage();
    $startTime = microtime(true);
    $queueKey = $this->helper->nDigitRandom(5);
    $this->logger->info("# " . $queueKey . " | (Starting " . $this->name . " Queue) | Batch : " . $batchSize);

    foreach ($queueItems as $item) {
      $i = 50;
      while ($i-- && $item->getStatus() == 'pending') {
        // if ($item->getStatus() != 'pending') {
        //   continue; // we do this to make sure this isn't being consumed by another parallel cron processes
        // }

        $error = false;
        $status = 'success';
        $errorMessage = '';

        try {
          $item->setStatus('processing')->save();
          $result = $this->processQueueItem($item);
          sleep(1);
          $totalProcessed++;
        } catch (\Throwable $e) {
          $error = true;
          $errorMessage = $e->getMessage();
          $this->logger->error('------------------------' . $this->name);
          $this->logger->error('Queue ID# ' . $item->getId());
          $this->logger->error($e->getMessage());
          $status = 'error';
          if ($e->getMessage() == 'The requested qty is not available') {
            $status = 'pending';
          } else if ($e->getMessage() == 'The requested sku is not assigned to given stock') {
            $status = 'pending';
          } else if (str_starts_with($e->getMessage(), 'No such entity with email')){
            $status = 'pending';
          } else if (str_starts_with($e->getMessage(), 'SQLSTATE[40001]: Serialization failure: 1213 Deadlock found when trying to get lock;')){
            $status = 'pending';
          } else if ($e->getMessage() == 'Product that you are trying to add is not available.') {
            $status = 'pending';
          }
          if ($status == 'error' && ($e->getMessage() != 'Email is required.')) {
            $this->logger->error($e->getTraceAsString());
          }
        }

        try {
          if ($error) {
            $item->setMessages($errorMessage);
          } else if (isset($result['data'])) {
            $item->setMessages(($result['is_new'] ?? false ? '' : 'Existing ') . 'ID: ' . $result['data']->getId() ?? '');
          }
          $item->setStatus($status);
          $item->save();
        } catch (\Throwable $e) {
          $this->logger->error('Failed update queue item status: ' . $e->getTraceAsString());
        }

      }
    }

    $memoryEnd = memory_get_usage();
    $endTime = microtime(true);
    $timeSpent = $this->helper->secondsToTime($endTime - $startTime);
    $memoryUsed = $this->helper->formatBytes($memoryEnd - $memoryStart);
    $this->logger->info("[$this->name" . "QueueManager] de-queue, # " . $queueKey . " | (Ending " . $this->name . " Queue) | Batched: " . $batchSize . " | # Processed: " . $totalProcessed . " | Memory Used: " . $memoryUsed . ' | Time: ' . $timeSpent);

    return $totalProcessed;
  }

  /**
   * Get next queue items to process
   *
   * @param int $batchSize
   * @return array
   */
  abstract public function dequeue($batchSize = 50);

  /**
   * Process queue item
   *
   * @param $queueItem
   * @return mixed
   */
  abstract public function processQueueItem($queueItem);

}