[Improvement]: Improve processing of generic data index queue (#344)

* Improved index queue processing

* Added migration

* Apply php-cs-fixer changes

* Async adding items for update/add operation

* Apply php-cs-fixer changes

* Removed auto-generated stub

* Apply php-cs-fixer changes

* Improved sql statements, avoided bc break

* Added message to transport, improved reading values from sql result

* Apply php-cs-fixer changes

* Reverted changes to generateSelectQuery

* Apply php-cs-fixer changes

* Add processing of messages to tests

* Using consume method

* Revert "Using consume method"

This reverts commit 139a547.

* Revert "Add processing of messages to tests"

This reverts commit 9dd4285.

* Implemented SynchronousProcessingRelatedIdsService

* Apply php-cs-fixer changes

* Refactor deleteQueueEntries

* Apply php-cs-fixer changes

* Added documentation

* Added upgrade note

* Changing runners for codeception tests

* Reverting to default runners

* Adapted installer to create new table layout

* Apply php-cs-fixer changes

* Cast id to string for compatibility with lower versions

---------

Co-authored-by: mcop1 <[email protected]>

Author: mcop1

config/pimcore/messenger.yaml

@@ -12,6 +12,7 @@ framework:
             Pimcore\Bundle\GenericDataIndexBundle\Message\DispatchQueueMessagesMessage: pimcore_generic_data_index_queue
             Pimcore\Bundle\GenericDataIndexBundle\Message\UpdateLanguageSettingsMessage: pimcore_generic_data_index_queue
             Pimcore\Bundle\GenericDataIndexBundle\Message\UpdateClassMappingMessage: pimcore_generic_data_index_queue
+            Pimcore\Bundle\GenericDataIndexBundle\Message\EnqueueRelatedIdsMessage: pimcore_generic_data_index_queue
         buses:
             messenger.bus.pimcore-generic-data-index:
                 middleware:

config/services/search/index.yaml

@@ -24,6 +24,9 @@ services:
     Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueue\SynchronousProcessingServiceInterface:
         class: Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueue\SynchronousProcessingService
 
+    Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueue\SynchronousProcessingRelatedIdsServiceInterface:
+        class: Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueue\SynchronousProcessingRelatedIdsService
+
     Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueue\EnqueueServiceInterface:
         class: Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueue\EnqueueService
 

doc/01_Installation/02_Upgrade.md

@@ -3,7 +3,8 @@
 Following steps are necessary during updating to newer versions.
 
 ## Upgrade to 2.2.0
-- Added `trackTotalHits` parameter to `DefaultSearchService` and `SearchExecutionService`. The default value is true,       which means that total hits will always be computed accurately, even if they exceed the search engines threshold for accurate hit calculation. Change this parameter to `null`, to use the default threshold, pass an integer value to set a specific one.
+- [Indexing] Added `id` column as new primary key to `generic_data_index_queue`. Please make sure to execute migrations.
+- [Searching] Added `trackTotalHits` parameter to `DefaultSearchService` and `SearchExecutionService`. The default value is true,       which means that total hits will always be computed accurately, even if they exceed the search engines threshold for accurate hit calculation. Change this parameter to `null`, to use the default threshold, pass an integer value to set a specific one.
 
 ## Upgrade to 2.1.0
 - Added support for Symfony 7

doc/02_Configuration/03_Index_Management.md

@@ -69,6 +69,27 @@ pimcore_generic_data_index:
             max_batch_size: 400
 ```
 
+#### Related elements
+
+The indexing queue is automatically populated whenever an element undergoes an update operation. This process includes not only the modified element itself but also any related elements. By default, this indexing occurs asynchronously through Symfony Messenger.
+
+For scenarios requiring immediate processing, you can temporarily switch to synchronous mode by utilizing the `SynchronousProcessingRelatedIdsServiceInterface`.
+
+Available methods are:
+
+| Method | Description |
+|--------|-------------|
+| `enable()` | Activates synchronous processing mode |
+| `disable()` | Reverts to asynchronous processing mode |
+| `isEnabled()` | Returns the current processing mode status |
+
+:::info
+
+Currently the `SynchronousProcessingRelatedIdsServiceInterface` interface does not influence the behavior of delete operations. They are always processed synchronously.
+
+:::
+
+
 ### Repairing Indices
 
 Sometimes it might be needed to delete and recreate the index from the Pimcore database

src/Entity/IndexQueue.php

@@ -29,10 +29,12 @@ class IndexQueue
     public const TABLE = 'generic_data_index_queue';
 
     #[ORM\Id]
+    #[ORM\Column(type: 'bigint')]
+    private int $id;
+
     #[ORM\Column(type: 'integer', options: ['unsigned' => true])]
     private int $elementId;
 
-    #[ORM\Id]
     #[ORM\Column(type: 'string', length: 20)]
     private string $elementType;
 
@@ -48,6 +50,18 @@ class IndexQueue
     #[ORM\Column(type: 'bigint', options: ['unsigned' => true, 'default' => 0])]
     private string $dispatched;
 
+    public function getId(): int
+    {
+        return $this->id;
+    }
+
+    public function setId(int $id): IndexQueue
+    {
+        $this->id = $id;
+
+        return $this;
+    }
+
     public function getElementId(): int
     {
         return $this->elementId;

src/EventSubscriber/AssetIndexUpdateSubscriber.php

@@ -16,6 +16,7 @@
 use Pimcore\Bundle\GenericDataIndexBundle\Enum\SearchIndex\IndexQueueOperation;
 use Pimcore\Bundle\GenericDataIndexBundle\Installer;
 use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueue\QueueMessagesDispatcher;
+use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueue\SynchronousProcessingRelatedIdsServiceInterface;
 use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueue\SynchronousProcessingServiceInterface;
 use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueueServiceInterface;
 use Pimcore\Event\AssetEvents;
@@ -31,7 +32,8 @@ public function __construct(
         private IndexQueueServiceInterface $indexQueueService,
         private Installer $installer,
         private QueueMessagesDispatcher $queueMessagesDispatcher,
-        private SynchronousProcessingServiceInterface $synchronousProcessing
+        private SynchronousProcessingServiceInterface $synchronousProcessing,
+        private SynchronousProcessingRelatedIdsServiceInterface $synchronousProcessingRelatedIds
     ) {
     }
 
@@ -54,7 +56,8 @@ public function updateAsset(AssetEvent $event): void
             ->updateIndexQueue(
                 element: $event->getAsset(),
                 operation: IndexQueueOperation::UPDATE->value,
-                processSynchronously: $this->synchronousProcessing->isEnabled()
+                processSynchronously: $this->synchronousProcessing->isEnabled(),
+                enqueueRelatedItemsAsync: $this->synchronousProcessingRelatedIds->isEnabled() === false
             )
             ->commit();
 

src/EventSubscriber/DataObjectIndexUpdateSubscriber.php

@@ -16,12 +16,13 @@
 use Pimcore\Bundle\GenericDataIndexBundle\Enum\SearchIndex\IndexQueueOperation;
 use Pimcore\Bundle\GenericDataIndexBundle\Installer;
 use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueue\QueueMessagesDispatcher;
+use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueue\SynchronousProcessingRelatedIdsServiceInterface;
 use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueue\SynchronousProcessingServiceInterface;
 use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueueServiceInterface;
 use Pimcore\Bundle\GenericDataIndexBundle\Traits\LoggerAwareTrait;
 use Pimcore\Event\DataObjectEvents;
 use Pimcore\Event\Model\DataObjectEvent;
-use Pimcore\Model\DataObject\AbstractObject;
+use Pimcore\Model\DataObject\Service;
 use Symfony\Component\EventDispatcher\EventSubscriberInterface;
 
 /**
@@ -35,7 +36,8 @@ public function __construct(
         private readonly Installer $installer,
         private readonly IndexQueueServiceInterface $indexQueueService,
         private readonly QueueMessagesDispatcher $queueMessagesDispatcher,
-        private readonly SynchronousProcessingServiceInterface $synchronousProcessing
+        private readonly SynchronousProcessingServiceInterface $synchronousProcessing,
+        private readonly SynchronousProcessingRelatedIdsServiceInterface $synchronousProcessingRelatedIds
     ) {
     }
 
@@ -62,19 +64,20 @@ public function updateDataObject(DataObjectEvent $event): void
             return;
         }
 
-        $inheritanceBackup = AbstractObject::getGetInheritedValues();
-        AbstractObject::setGetInheritedValues(true);
+        $dataObject = $event->getObject();
+        Service::useInheritedValues(true, fn () =>
+            $this->indexQueueService
+                ->updateIndexQueue(
+                    $dataObject,
+                    IndexQueueOperation::UPDATE->value,
+                    $this->synchronousProcessing->isEnabled(),
+                    $dataObject->hasChildren(includingUnpublished: true),
+                    $this->synchronousProcessingRelatedIds->isEnabled() === false
+                )
+                ->commit()
+        );
 
-        $this->indexQueueService
-            ->updateIndexQueue(
-                element: $event->getObject(),
-                operation: IndexQueueOperation::UPDATE->value,
-                processSynchronously: $this->synchronousProcessing->isEnabled()
-            )
-            ->commit();
         $this->queueMessagesDispatcher->dispatchQueueMessages();
-
-        AbstractObject::setGetInheritedValues($inheritanceBackup);
     }
 
     public function deleteDataObject(DataObjectEvent $event): void
@@ -83,11 +86,13 @@ public function deleteDataObject(DataObjectEvent $event): void
             return;
         }
 
+        $dataObject = $event->getObject();
         $this->indexQueueService
             ->updateIndexQueue(
-                element: $event->getObject(),
-                operation: IndexQueueOperation::DELETE->value,
-                processSynchronously: $this->synchronousProcessing->isEnabled()
+                $dataObject,
+                IndexQueueOperation::DELETE->value,
+                $this->synchronousProcessing->isEnabled(),
+                $dataObject->hasChildren(includingUnpublished: true)
             )
             ->commit();
         $this->queueMessagesDispatcher->dispatchQueueMessages();

src/EventSubscriber/DocumentIndexUpdateSubscriber.php

@@ -16,6 +16,7 @@
 use Pimcore\Bundle\GenericDataIndexBundle\Enum\SearchIndex\IndexQueueOperation;
 use Pimcore\Bundle\GenericDataIndexBundle\Installer;
 use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueue\QueueMessagesDispatcher;
+use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueue\SynchronousProcessingRelatedIdsServiceInterface;
 use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueue\SynchronousProcessingServiceInterface;
 use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueueServiceInterface;
 use Pimcore\Event\DocumentEvents;
@@ -31,7 +32,8 @@ public function __construct(
         private IndexQueueServiceInterface $indexQueueService,
         private Installer $installer,
         private QueueMessagesDispatcher $queueMessagesDispatcher,
-        private SynchronousProcessingServiceInterface $synchronousProcessing
+        private SynchronousProcessingServiceInterface $synchronousProcessing,
+        private SynchronousProcessingRelatedIdsServiceInterface $synchronousProcessingRelatedIds
     ) {
     }
 
@@ -54,7 +56,8 @@ public function updateDocument(DocumentEvent $event): void
             ->updateIndexQueue(
                 element: $event->getDocument(),
                 operation: IndexQueueOperation::UPDATE->value,
-                processSynchronously: $this->synchronousProcessing->isEnabled()
+                processSynchronously: $this->synchronousProcessing->isEnabled(),
+                enqueueRelatedItemsAsync: $this->synchronousProcessingRelatedIds->isEnabled() === false
             )
             ->commit();
 

src/Installer.php

@@ -21,7 +21,7 @@
 use Doctrine\DBAL\Schema\SchemaException;
 use Pimcore;
 use Pimcore\Bundle\GenericDataIndexBundle\Entity\IndexQueue;
-use Pimcore\Bundle\GenericDataIndexBundle\Migrations\Version20240325081139;
+use Pimcore\Bundle\GenericDataIndexBundle\Migrations\Version20251009110653;
 use Pimcore\Extension\Bundle\Installer\Exception\InstallationException;
 use Symfony\Component\HttpKernel\Bundle\BundleInterface;
 
@@ -40,7 +40,7 @@ public function __construct(
 
     public function getLastMigrationVersionClassName(): ?string
     {
-        return Version20240325081139::class;
+        return Version20251009110653::class;
     }
 
     /**
@@ -89,6 +89,7 @@ private function installIndexQueueTable(Schema $schema): void
     {
         if (!$schema->hasTable(IndexQueue::TABLE)) {
             $queueTable = $schema->createTable(IndexQueue::TABLE);
+            $queueTable->addColumn('id', 'bigint', ['autoincrement' => true]);
             $queueTable->addColumn('elementId', 'integer', ['notnull' => true, 'unsigned' => true]);
             $queueTable->addColumn('elementType', 'string', ['notnull' => true, 'length' => 20]);
             $queueTable->addColumn('elementIndexName', 'string', ['notnull' => true, 'length' => 255]);
@@ -100,7 +101,8 @@ private function installIndexQueueTable(Schema $schema): void
                 'default' => 0,
             ]);
 
-            $queueTable->setPrimaryKey(['elementId', 'elementType']);
+            $queueTable->setPrimaryKey(['id']);
+            $queueTable->addUniqueIndex(['elementId', 'elementType'], IndexQueue::TABLE . '_element_id_type');
             $queueTable->addIndex(['dispatched'], IndexQueue::TABLE . '_dispatched');
             $queueTable->addIndex(['operationTime'], IndexQueue::TABLE . '_operation_time');
         }

src/Message/EnqueueRelatedIdsMessage.php

@@ -0,0 +1,50 @@
+<?php
+declare(strict_types=1);
+
+/**
+ * This source file is available under the terms of the
+ * Pimcore Open Core License (POCL)
+ * Full copyright and license information is available in
+ * LICENSE.md which is distributed with this source code.
+ *
+ *  @copyright  Copyright (c) Pimcore GmbH (https://www.pimcore.com)
+ *  @license    Pimcore Open Core License (POCL)
+ */
+
+namespace Pimcore\Bundle\GenericDataIndexBundle\Message;
+
+use Pimcore\Bundle\GenericDataIndexBundle\Enum\SearchIndex\ElementType;
+
+/**
+ * @internal
+ */
+final readonly class EnqueueRelatedIdsMessage
+{
+    public function __construct(
+        private int $elementId,
+        private ElementType $elementType,
+        private string $operation,
+        private bool $addParentElement
+    ) {
+    }
+
+    public function getElementId(): int
+    {
+        return $this->elementId;
+    }
+
+    public function getElementType(): ElementType
+    {
+        return $this->elementType;
+    }
+
+    public function getOperation(): string
+    {
+        return $this->operation;
+    }
+
+    public function getAddParentElement(): bool
+    {
+        return $this->addParentElement;
+    }
+}