src/Elasticsearch/Admin/AdminSearchRegistry.php line 136

  1. <?php declare(strict_types=1);
  3. namespace Shopware\Elasticsearch\Admin;
  5. use Doctrine\DBAL\Connection;
  6. use Doctrine\DBAL\Exception;
  7. use OpenSearch\Client;
  8. use Shopware\Core\Framework\DataAbstractionLayer\Event\EntityWrittenContainerEvent;
  9. use Shopware\Core\Framework\Event\ProgressAdvancedEvent;
  10. use Shopware\Core\Framework\Event\ProgressFinishedEvent;
  11. use Shopware\Core\Framework\Event\ProgressStartedEvent;
  12. use Shopware\Core\Framework\Log\Package;
  13. use Shopware\Core\Framework\Uuid\Uuid;
  14. use Shopware\Elasticsearch\Admin\Indexer\AbstractAdminIndexer;
  15. use Shopware\Elasticsearch\Exception\ElasticsearchIndexingException;
  16. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  17. use Symfony\Component\Messenger\Handler\MessageHandlerInterface;
  18. use Symfony\Component\Messenger\MessageBusInterface;
  19. use Symfony\Contracts\EventDispatcher\EventDispatcherInterface;
  21. /**
  22.  * @internal
  23.  * @final
  24.  */
  25. #[Package('system-settings')]
  26. class AdminSearchRegistry implements MessageHandlerInterface, EventSubscriberInterface
  27. {
  28.     /**
  29.      * @var array<string, mixed>
  30.      */
  31.     private readonly array $indexer;
  33.     /**
  34.      * @var array<mixed>
  35.      */
  36.     private readonly array $config;
  38.     /**
  39.      * @param AbstractAdminIndexer[] $indexer
  40.      * @param array<mixed> $config
  41.      * @param array<mixed> $mapping
  42.      */
  43.     public function __construct(
  44.         $indexer,
  45.         private readonly Connection $connection,
  46.         private readonly MessageBusInterface $queue,
  47.         private readonly EventDispatcherInterface $dispatcher,
  48.         private readonly Client $client,
  49.         private readonly AdminElasticsearchHelper $adminEsHelper,
  50.         array $config,
  51.         private readonly array $mapping
  52.     ) {
  53.         $this->indexer = $indexer instanceof \Traversable ? iterator_to_array($indexer) : $indexer;
  55.         if (isset($config['settings']['index'])) {
  56.             if (\array_key_exists('number_of_shards', $config['settings']['index']) && $config['settings']['index']['number_of_shards'] === null) {
  57.                 unset($config['settings']['index']['number_of_shards']);
  58.             }
  60.             if (\array_key_exists('number_of_replicas', $config['settings']['index']) && $config['settings']['index']['number_of_replicas'] === null) {
  61.                 unset($config['settings']['index']['number_of_replicas']);
  62.             }
  63.         }
  65.         $this->config = $config;
  66.     }
  68.     public function __invoke(AdminSearchIndexingMessage $message): void
  69.     {
  70.         $indexer = $this->getIndexer($message->getEntity());
  72.         $documents = $indexer->fetch($message->getIds());
  74.         $this->push($indexer, $message->getIndices(), $documents, $message->getIds());
  75.     }
  77.     public static function getSubscribedEvents(): array
  78.     {
  79.         return [
  80.             EntityWrittenContainerEvent::class => [
  81.                 ['refresh', -1000],
  82.             ],
  83.         ];
  84.     }
  86.     /**
  87.      * @return iterable<class-string>
  88.      */
  89.     public static function getHandledMessages(): iterable
  90.     {
  91.         return [
  92.             AdminSearchIndexingMessage::class,
  93.         ];
  94.     }
  96.     public function iterate(AdminIndexingBehavior $indexingBehavior): void
  97.     {
  98.         if ($this->adminEsHelper->getEnabled() === false) {
  99.             return;
  100.         }
  102.         /** @var array<string> $entities */
  103.         $entities = array_keys($this->indexer);
  105.         if ($indexingBehavior->getOnlyEntities()) {
  106.             $entities = array_intersect($entities, $indexingBehavior->getOnlyEntities());
  107.         } elseif ($indexingBehavior->getSkipEntities()) {
  108.             $entities = array_diff($entities, $indexingBehavior->getSkipEntities());
  109.         }
  111.         $indices = $this->createIndices($entities);
  113.         foreach ($entities as $entityName) {
  114.             $indexer = $this->getIndexer($entityName);
  115.             $iterator = $indexer->getIterator();
  117.             $this->dispatcher->dispatch(new ProgressStartedEvent($indexer->getName(), $iterator->fetchCount()));
  119.             while ($ids = $iterator->fetch()) {
  120.                 // we provide no queue when the data is sent by the admin
  121.                 if ($indexingBehavior->getNoQueue() === true) {
  122.                     $this->__invoke(new AdminSearchIndexingMessage($indexer->getEntity(), $indexer->getName(), $indices, $ids));
  123.                 } else {
  124.                     $this->queue->dispatch(new AdminSearchIndexingMessage($indexer->getEntity(), $indexer->getName(), $indices, $ids));
  125.                 }
  127.                 $this->dispatcher->dispatch(new ProgressAdvancedEvent(\count($ids)));
  128.             }
  130.             $this->dispatcher->dispatch(new ProgressFinishedEvent($indexer->getName()));
  131.         }
  133.         $this->swapAlias($indices);
  134.     }
  136.     public function refresh(EntityWrittenContainerEvent $event): void
  137.     {
  138.         if ($this->adminEsHelper->getEnabled() === false || !$this->isIndexedEntityWritten($event)) {
  139.             return;
  140.         }
  142.         if ($this->adminEsHelper->getRefreshIndices()) {
  143.             $this->refreshIndices();
  144.         }
  146.         /** @var array<string, string> $indices */
  147.         $indices = $this->connection->fetchAllKeyValue('SELECT `alias`, `index` FROM admin_elasticsearch_index_task');
  149.         if (empty($indices)) {
  150.             return;
  151.         }
  153.         foreach ($this->indexer as $indexer) {
  154.             $ids = $event->getPrimaryKeys($indexer->getEntity());
  156.             if (empty($ids)) {
  157.                 continue;
  158.             }
  159.             $documents = $indexer->fetch($ids);
  161.             $this->push($indexer, $indices, $documents, $ids);
  162.         }
  163.     }
  165.     /**
  166.      * @return AbstractAdminIndexer[]
  167.      */
  168.     public function getIndexers(): iterable
  169.     {
  170.         return $this->indexer;
  171.     }
  173.     public function getIndexer(string $name): AbstractAdminIndexer
  174.     {
  175.         $indexer = $this->indexer[$name] ?? null;
  176.         if ($indexer) {
  177.             return $indexer;
  178.         }
  180.         throw new ElasticsearchIndexingException([\sprintf('Indexer for name %s not found', $name)]);
  181.     }
  183.     private function isIndexedEntityWritten(EntityWrittenContainerEvent $event): bool
  184.     {
  185.         foreach ($this->indexer as $indexer) {
  186.             $ids = $event->getPrimaryKeys($indexer->getEntity());
  188.             if (!empty($ids)) {
  189.                 return true;
  190.             }
  191.         }
  193.         return false;
  194.     }
  196.     /**
  197.      * @param array<string, string> $indices
  198.      * @param array<string, array<string|int, string>> $data
  199.      * @param array<string> $ids
  200.      */
  201.     private function push(AbstractAdminIndexer $indexer, array $indices, array $data, array $ids): void
  202.     {
  203.         $alias = $this->adminEsHelper->getIndex($indexer->getName());
  205.         if (!isset($indices[$alias])) {
  206.             return;
  207.         }
  209.         $toRemove = array_filter($ids, static fn (string $id): bool => !isset($data[$id]));
  211.         $documents = [];
  212.         foreach ($data as $id => $document) {
  213.             $documents[] = ['index' => ['_id' => $id]];
  215.             $documents[] = \array_replace(
  216.                 ['entityName' => $indexer->getEntity(), 'parameters' => [], 'textBoosted' => '', 'text' => ''],
  217.                 $document
  218.             );
  219.         }
  221.         foreach ($toRemove as $id) {
  222.             $documents[] = ['delete' => ['_id' => $id]];
  223.         }
  225.         $arguments = [
  226.             'index' => $indices[$alias],
  227.             'body' => $documents,
  228.         ];
  230.         $result = $this->client->bulk($arguments);
  232.         if (\is_array($result) && !empty($result['errors'])) {
  233.             $errors = $this->parseErrors($result);
  235.             throw new ElasticsearchIndexingException($errors);
  236.         }
  237.     }
  239.     /**
  240.      * @param array<string> $entities
  241.      *
  242.      * @throws Exception
  243.      *
  244.      * @return array<string, string>
  245.      */
  246.     private function createIndices(array $entities): array
  247.     {
  248.         $indexTasks = [];
  249.         $indices = [];
  250.         foreach ($entities as $entityName) {
  251.             $indexer = $this->getIndexer($entityName);
  252.             $alias = $this->adminEsHelper->getIndex($indexer->getName());
  253.             $index = $alias . '_' . time();
  255.             if ($this->indexExists($index)) {
  256.                 continue;
  257.             }
  259.             $indices[$alias] = $index;
  261.             $this->create($indexer, $index, $alias);
  263.             $iterator = $indexer->getIterator();
  264.             $indexTasks[] = [
  265.                 'id' => Uuid::randomBytes(),
  266.                 '`entity`' => $indexer->getEntity(),
  267.                 '`index`' => $index,
  268.                 '`alias`' => $alias,
  269.                 '`doc_count`' => $iterator->fetchCount(),
  270.             ];
  271.         }
  273.         $this->connection->executeStatement(
  274.             'DELETE FROM admin_elasticsearch_index_task WHERE `entity` IN (:entities)',
  275.             ['entities' => $entities],
  276.             ['entities' => Connection::PARAM_STR_ARRAY]
  277.         );
  279.         foreach ($indexTasks as $task) {
  280.             $this->connection->insert('admin_elasticsearch_index_task', $task);
  281.         }
  283.         return $indices;
  284.     }
  286.     private function refreshIndices(): void
  287.     {
  288.         $entities = [];
  289.         $indexTasks = [];
  290.         foreach ($this->indexer as $indexer) {
  291.             $alias = $this->adminEsHelper->getIndex($indexer->getName());
  293.             if ($this->aliasExists($alias)) {
  294.                 continue;
  295.             }
  297.             $index = $alias . '_' . time();
  298.             $this->create($indexer, $index, $alias);
  300.             $entities[] = $indexer->getEntity();
  302.             $iterator = $indexer->getIterator();
  303.             $indexTasks[] = [
  304.                 'id' => Uuid::randomBytes(),
  305.                 '`entity`' => $indexer->getEntity(),
  306.                 '`index`' => $index,
  307.                 '`alias`' => $alias,
  308.                 '`doc_count`' => $iterator->fetchCount(),
  309.             ];
  310.         }
  312.         $this->connection->executeStatement(
  313.             'DELETE FROM admin_elasticsearch_index_task WHERE `entity` IN (:entities)',
  314.             ['entities' => $entities],
  315.             ['entities' => Connection::PARAM_STR_ARRAY]
  316.         );
  318.         foreach ($indexTasks as $task) {
  319.             $this->connection->insert('admin_elasticsearch_index_task', $task);
  320.         }
  321.     }
  323.     private function create(AbstractAdminIndexer $indexer, string $index, string $alias): void
  324.     {
  325.         $mapping = $indexer->mapping([
  326.             'properties' => [
  327.                 'id' => ['type' => 'keyword'],
  328.                 'textBoosted' => ['type' => 'text'],
  329.                 'text' => ['type' => 'text'],
  330.                 'entityName' => ['type' => 'keyword'],
  331.                 'parameters' => ['type' => 'keyword'],
  332.             ],
  333.         ]);
  335.         $mapping = array_merge_recursive($mapping, $this->mapping);
  337.         $body = array_merge(
  338.             $this->config,
  339.             ['mappings' => $mapping]
  340.         );
  342.         $this->client->indices()->create([
  343.             'index' => $index,
  344.             'body' => $body,
  345.         ]);
  347.         $this->createAliasIfNotExisting($index, $alias);
  348.     }
  350.     private function indexExists(string $name): bool
  351.     {
  352.         return $this->client->indices()->exists(['index' => $name]);
  353.     }
  355.     private function aliasExists(string $alias): bool
  356.     {
  357.         return $this->client->indices()->existsAlias(['name' => $alias]);
  358.     }
  360.     /**
  361.      * @param array<string, array<array<string, mixed>>> $result
  362.      *
  363.      * @return array<array{reason: string}|string>
  364.      */
  365.     private function parseErrors(array $result): array
  366.     {
  367.         $errors = [];
  368.         foreach ($result['items'] as $item) {
  369.             $item = $item['index'] ?? $item['delete'];
  371.             if (\in_array($item['status'], [200, 201], true)) {
  372.                 continue;
  373.             }
  375.             $errors[] = [
  376.                 'index' => $item['_index'],
  377.                 'id' => $item['_id'],
  378.                 'type' => $item['error']['type'] ?? $item['_type'],
  379.                 'reason' => $item['error']['reason'] ?? $item['result'],
  380.             ];
  381.         }
  383.         return $errors;
  384.     }
  386.     private function createAliasIfNotExisting(string $index, string $alias): void
  387.     {
  388.         $exist = $this->client->indices()->existsAlias(['name' => $alias]);
  390.         if ($exist) {
  391.             return;
  392.         }
  394.         $this->putAlias($index, $alias);
  395.     }
  397.     /**
  398.      * @param array<string, string> $indices
  399.      */
  400.     private function swapAlias($indices): void
  401.     {
  402.         foreach ($indices as $alias => $index) {
  403.             $exist = $this->client->indices()->existsAlias(['name' => $alias]);
  405.             if (!$exist) {
  406.                 $this->putAlias($index, $alias);
  408.                 return;
  409.             }
  411.             $current = $this->client->indices()->getAlias(['name' => $alias]);
  413.             if (!isset($current[$index])) {
  414.                 $this->putAlias($index, $alias);
  415.             }
  417.             unset($current[$index]);
  418.             $current = array_keys($current);
  420.             foreach ($current as $value) {
  421.                 $this->client->indices()->delete(['index' => $value]);
  422.             }
  423.         }
  424.     }
  426.     private function putAlias(string $index, string $alias): void
  427.     {
  428.         $this->client->indices()->refresh([
  429.             'index' => $index,
  430.         ]);
  431.         $this->client->indices()->putAlias(['index' => $index, 'name' => $alias]);
  432.     }
  433. }