Add feature product data fetch with queue

Api/WebhookManagementInterface.php

@@ -0,0 +1,14 @@
+<?php
+
+namespace Lipscore\RatingsReviews\Api;
+
+use Magento\Framework\Exception\LocalizedException;
+
+interface WebhookManagementInterface
+{
+    /**
+     * @return bool
+     * @throws LocalizedException
+     */
+    public function handle(): bool;
+}

Block/Product/JsonLd.php

@@ -0,0 +1,56 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Lipscore\RatingsReviews\Block\Product;
+
+use Lipscore\RatingsReviews\Model\Service\JsonLd\LipscoreReadService;
+use Magento\Framework\Registry;
+use Magento\Framework\View\Element\Template;
+
+class JsonLd extends Template
+{
+    protected $registry;
+
+    protected $readService;
+
+    public function __construct(
+        Template\Context $context,
+        Registry $registry,
+        LipscoreReadService $readService,
+        array $data = []
+    ) {
+        $this->registry = $registry;
+        $this->readService = $readService;
+
+        parent::__construct($context, $data);
+    }
+
+    /**
+     * @return \Magento\Catalog\Model\Product|null
+     */
+    public function getProduct()
+    {
+        return $this->registry->registry('current_product');
+    }
+
+    public function getProductJsonLd(): ?array
+    {
+        $product = $this->getProduct();
+        if (!$product) {
+            return null;
+        }
+
+        return $this->readService->getJsonLdProduct($product);
+    }
+
+    public function getReviewsJsonLd(): ?array
+    {
+        $product = $this->getProduct();
+        if (!$product) {
+            return null;
+        }
+
+        return $this->readService->getJsonLdProductReviews($product);
+    }
+}

Console/Command/DeleteAllHooks.php

@@ -0,0 +1,39 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Lipscore\RatingsReviews\Console\Command;
+
+use Lipscore\RatingsReviews\Model\ApiKeyRepository;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+class DeleteAllHooks extends Command
+{
+    protected $repository;
+
+    public function __construct(
+        ApiKeyRepository $repository,
+        ?string $name = null
+    ) {
+        $this->repository = $repository;
+
+        parent::__construct($name);
+    }
+
+    protected function configure(): void
+    {
+        $this->setName('lipscore:hooks:delete')
+            ->setDescription('Deletes Lipscore API hooks');
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output)
+    {
+        $this->repository->deleteHooksForAllKeys();
+
+        $output->writeln("<info>Lipscore Hooks Delete finished</info>");
+
+        return 0;
+    }
+}

Console/Command/ProcessQueue.php

@@ -0,0 +1,44 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Lipscore\RatingsReviews\Console\Command;
+
+use Lipscore\RatingsReviews\Model\Action\ProcessLipscoreQueue as Action;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+class ProcessQueue extends Command
+{
+    protected $action;
+
+    public function __construct(
+        Action $action,
+        ?string $name = null
+    ) {
+        $this->action = $action;
+
+        parent::__construct($name);
+    }
+
+    protected function configure(): void
+    {
+        $this->setName('lipscore:queue')
+            ->setDescription('Asks for reviews data if product is present in queue table');
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        try {
+            $this->action->execute();
+            $output->writeln('<info>Lipscore Queue Finished</info>');
+
+            return Command::SUCCESS;
+        } catch (\Throwable $e) {
+            $output->writeln('<error>Lipscore Queue Failed</error>');
+
+            return Command::FAILURE;
+        }
+    }
+}

Console/Command/SyncHooks.php

@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Lipscore\RatingsReviews\Console\Command;
+
+use Lipscore\RatingsReviews\Model\ApiKeyRepository;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+class SyncHooks extends Command
+{
+    protected $repository;
+
+    public function __construct(
+        ApiKeyRepository $repository,
+        ?string $name = null
+    ) {
+        $this->repository = $repository;
+
+        parent::__construct($name);
+    }
+
+    protected function configure(): void
+    {
+        $this->setName('lipscore:hooks:sync')
+            ->setDescription('Asks Lipscore API for product refresh');
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output)
+    {
+        $this->repository->syncKeys();
+        $output->writeln("<info>Lipscore Sync Finished</info>");
+
+        return 0;
+    }
+}

Cron/ProcessQueue.php

@@ -0,0 +1,32 @@
+<?php
+
+namespace Lipscore\RatingsReviews\Cron;
+
+use Lipscore\RatingsReviews\Model\Action\ProcessLipscoreQueue as Action;
+use Lipscore\RatingsReviews\Model\QueueRepository;
+use Magento\Cron\Model\Schedule;
+
+class ProcessQueue
+{
+    protected $action;
+
+    protected $queueRepository;
+
+    public function __construct(
+        Action $action,
+        QueueRepository $queueRepository
+    ) {
+        $this->action = $action;
+        $this->queueRepository = $queueRepository;
+    }
+
+    public function execute(Schedule $schedule): void
+    {
+        $this->action->execute();
+    }
+
+    public function cleanup(Schedule $schedule): void
+    {
+        $this->queueRepository->cleanup();
+    }
+}

Exception/ApiException.php

@@ -0,0 +1,10 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Lipscore\RatingsReviews\Exception;
+
+
+class ApiException extends \Exception
+{
+}

Model/Action/ProcessLipscoreQueue.php

@@ -0,0 +1,130 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Lipscore\RatingsReviews\Model\Action;
+
+use Lipscore\RatingsReviews\Model\ApiKeyRepository;
+use Lipscore\RatingsReviews\Model\QueueRepository;
+use Lipscore\RatingsReviews\Model\Service\JsonLd\LipscoreReadService;
+use Psr\Log\LoggerInterface;
+
+class ProcessLipscoreQueue
+{
+    private const BATCH_SIZE  = 20;
+    private const MAX_BATCHES = 10;
+
+    public function __construct(
+        private ApiKeyRepository $apiKeyRepository,
+        private QueueRepository $queueRepository,
+        private LipscoreReadService $readService,
+        private LoggerInterface $logger
+    ) {}
+
+    public function execute(): void
+    {
+        $start = microtime(true);
+        $keys = $this->apiKeyRepository->collectAllKeys(); // [id => apiKey]
+
+        try {
+            $apiKeys = $this->apiKeyRepository->buildKeyContextMap($keys);
+            if (!$apiKeys) {
+                $this->logger->warning('Lipscore: no API keys configured');
+                return;
+            }
+
+            $totalBatches = 0;
+            $totalItems   = 0;
+
+            foreach ($apiKeys as $context) {
+                [$batches, $items] = $this->processForKey($context);
+                $totalBatches += $batches;
+                $totalItems   += $items;
+            }
+
+            $this->logger->info('Lipscore queue complete', [
+                'keys'       => count($apiKeys),
+                'batches'    => $totalBatches,
+                'items'      => $totalItems,
+                'duration_s' => round(microtime(true) - $start, 2),
+            ]);
+
+        } catch (\Throwable $e) {
+            $this->logger->error('Lipscore queue cron error', [
+                'error_class' => get_class($e),
+                'message'     => $e->getMessage(),
+            ]);
+        }
+    }
+
+    private function processForKey(array $keyContext): array
+    {
+        $keyId = $this->apiKeyRepository->getIdByKey($keyContext['api_key']);
+        if (!$keyId) {
+            return [0, 0];
+        }
+
+        $keyContext['key_id'] = $keyId;
+
+        $batches = 0;
+        $total   = 0;
+
+        do {
+            $rows = $this->queueRepository->dequeue($keyId, self::BATCH_SIZE);
+            if (!$rows) {
+                break;
+            }
+
+            $productIds = array_map(
+                static fn(array $row): int => (int)$row['product_id'],
+                $rows
+            );
+
+            $this->logger->debug('Lipscore queue batch', [
+                'key_id'      => $keyId,
+                'product_ids' => $productIds,
+            ]);
+
+            try {
+                $result = $this->readService->fetchAndSaveBatch($productIds, $keyContext);
+                $saved  = $result['saved'] ?? [];
+
+                foreach ($rows as $row) {
+                    $queueId   = (int)$row['queue_id'];
+                    $productId = (int)$row['product_id'];
+                    $success   = in_array($productId, $saved, true);
+
+                    $this->queueRepository->markProcessed($queueId, $success);
+
+                    if (!$success) {
+                        $this->logger->warning('Lipscore: no data for product', [
+                            'product_id' => $productId,
+                        ]);
+                    }
+
+                    $total++;
+                }
+
+            } catch (\Throwable $e) {
+                foreach ($rows as $row) {
+                    $this->queueRepository->markProcessed(
+                        (int)$row['queue_id'],
+                        false
+                    );
+                }
+
+                $this->logger->error('Lipscore batch failed', [
+                    'api_key'     => substr($keyContext['api_key'], 0, 4) . '****',
+                    'error_class' => get_class($e),
+                    'error'       => $e->getMessage(),
+                ]);
+            }
+
+            $batches++;
+            usleep(100000); // 100ms throttle
+
+        } while ($batches < self::MAX_BATCHES);
+
+        return [$batches, $total];
+    }
+}