[Improvement]: Improve processing of generic data index queue

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

@@ -2,6 +2,9 @@
 
 Following steps are necessary during updating to newer versions.
 
+## Upgrade to 2.2.0
+- [Indexing] Added `id` column as new primary key to `generic_data_index_queue`. Please make sure to execute migrations.
+
 ## Upgrade to 2.1.0
 - Added support for Symfony 7
 - [Indexing] Added sort index for documents

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;
+    }
+}

src/MessageHandler/EnqueueRelatedIdsHandler.php

@@ -0,0 +1,62 @@
+<?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\MessageHandler;
+
+use Pimcore\Bundle\GenericDataIndexBundle\Enum\SearchIndex\ElementType;
+use Pimcore\Bundle\GenericDataIndexBundle\Message\EnqueueRelatedIdsMessage;
+use Pimcore\Bundle\GenericDataIndexBundle\Service\ElementServiceInterface;
+use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueue\QueueMessagesDispatcher;
+use Pimcore\Bundle\GenericDataIndexBundle\Service\SearchIndex\IndexQueueServiceInterface;
+use Pimcore\Model\DataObject\AbstractObject;
+use Symfony\Component\Messenger\Attribute\AsMessageHandler;
+
+/**
+ * @internal
+ */
+#[AsMessageHandler]
+final readonly class EnqueueRelatedIdsHandler
+{
+    public function __construct(
+        private IndexQueueServiceInterface $indexQueueService,
+        private QueueMessagesDispatcher $queueMessagesDispatcher,
+        private ElementServiceInterface $elementService
+    ) {
+    }
+
+    public function __invoke(EnqueueRelatedIdsMessage $message): void
+    {
+        $inheritanceBackup = null;
+        $elementType = $message->getElementType();
+        $element = $this->elementService->getElementByType($message->getElementId(), $elementType->value);
+
+        if ($elementType === ElementType::DATA_OBJECT) {
+            $inheritanceBackup = AbstractObject::getGetInheritedValues();
+            AbstractObject::setGetInheritedValues(true);
+        }
+
+        $this->indexQueueService->updateIndexQueue(
+            $element,
+            $message->getOperation(),
+            !$message->getAddParentElement(),
+            true,
+            false
+        )->commit();
+
+        if ($inheritanceBackup !== null) {
+            AbstractObject::setGetInheritedValues($inheritanceBackup);
+        }
+
+        $this->queueMessagesDispatcher->dispatchQueueMessages();
+    }
+}