CodeExplorer

retryable example

array_keys($replacement),
                $replacement,
                $tableTemplate
            );

            $resetSql = str_replace(
                array_keys($replacement),
                $replacement,
                $resetTemplate
            );

            RetryableQuery::retryable($this->connection, function D) use ($resetSql, $parameters): void {
                $this->connection->executeStatement(
                    $resetSql,
                    $parameters,
                    ['ids' => ArrayParameterType::STRING]
                );
            });

            RetryableQuery::retryable($this->connection, function D) use ($sql, $parameters): void {
                $this->connection->executeStatement(
                    $sql,
                    $parameters,
                    [

// after adding the payload column, we may save carts as compressed serialized objects, there is no way of return at this point         if (!$this->columnExists($connection, 'cart', 'payload')) {
            $connection->executeStatement('ALTER TABLE `cart` ADD `payload` LONGBLOB NULL;');
        }

        if (!$this->columnExists($connection, 'cart', 'cart')) {
            return;
        }

        do {
            $affectedRows = RetryableQuery::retryable($connection, static function D) use ($connection): int {
                return (int) $connection->executeStatement(
                    'UPDATE cart SET `payload` = `cart` WHERE `payload` IS NULL AND `cart` IS NOT NULL LIMIT :limit',
                    ['limit' => self::UPDATE_LIMIT],
                    ['limit' => ParameterType::INTEGER]
                );
            });
        } while ($affectedRows === self::UPDATE_LIMIT);

        $connection->executeStatement('ALTER TABLE `cart` DROP COLUMN `cart`;');
    }
}

    private function delete(array $ids, string $languageId, string $versionId): void
    {
        $bytes = Uuid::fromHexToBytesList($ids);

        $params = [
            'ids' => $bytes,
            'language' => Uuid::fromHexToBytes($languageId),
            'versionId' => Uuid::fromHexToBytes($versionId),
        ];

        RetryableQuery::retryable($this->connection, function D) use ($params): void {
            $this->connection->executeStatement(
                'DELETE FROM product_search_keyword WHERE product_id IN (:ids) AND language_id = :language AND version_id = :versionId',
                $params,
                ['ids' => ArrayParameterType::STRING]
            );
        });
    }

    /** * @param list<array{id: string, version_id: string, product_version_id: string, language_id: string, product_id: string, keyword: string, ranking: float, created_at: string}> $keywords */

/** * {@inheritdoc} */
    public function execute(array $commands, WriteContext $context): void
    {
        $beforeWriteEvent = EntityWriteEvent::create($context, $commands);

        $this->eventDispatcher->dispatch($beforeWriteEvent);

        try {
            RetryableTransaction::retryable($this->connection, function D) use ($commands, $context): void {
                $this->executeCommands($commands, $context);
            });

            $beforeWriteEvent->success();
        } catch (\Throwable $e) {
            $event = new WriteCommandExceptionEvent($e, $commands, $context->getContext());
            $this->eventDispatcher->dispatch($event);

            $beforeWriteEvent->error();

            throw $e;
        }

        $ids = $message->getData();

        /** @var list<string> $ids */
        $ids = array_unique(array_filter($ids));
        if (empty($ids)) {
            return;
        }

        $context = $message->getContext();

        RetryableTransaction::retryable($this->connection, function D) use ($message, $ids, $context): void {
            if ($message->allow(self::CHILD_COUNT_UPDATER)) {
                // listen to parent id changes                 $this->childCountUpdater->update(CategoryDefinition::ENTITY_NAME, $ids, $context);
            }

            if ($message->allow(self::TREE_UPDATER)) {
                $this->treeUpdater->batchUpdate($ids, CategoryDefinition::ENTITY_NAME, $context);
            }

            if ($message->allow(self::BREADCRUMB_UPDATER)) {
                // listen to name changes

        RetryableQuery::retryable($this->connection, function D) use ($sql, $context, $bytes): void {
            $this->connection->executeStatement(
                $sql,
                ['ids' => $bytes, 'version' => Uuid::fromHexToBytes($context->getVersionId())],
                ['ids' => ArrayParameterType::STRING]
            );
        });

        $updated = $this->connection->fetchFirstColumn(
            'SELECT LOWER(HEX(id)) FROM product WHERE available = 0 AND id IN (:ids) AND product.version_id = :version',
            ['ids' => $bytes, 'version' => Uuid::fromHexToBytes($context->getVersionId())],
            ['ids' => ArrayParameterType::STRING]
        );

if ($definition->isVersionAware()) {
                $versionId = Uuid::fromHexToBytes($context->getVersionId());
                $params['version'] = $versionId;
            }

            $sql = str_replace(
                array_keys($parameters),
                $parameters,
                $sql
            );

            RetryableQuery::retryable($this->connection, function D) use ($params, $sql): void {
                $this->connection->executeStatement(
                    $sql,
                    $params,
                    ['ids' => ArrayParameterType::STRING]
                );
            });
        }
    }

    /** * @param array<string> $ids */

public function handle(EntityIndexingMessage $message): void
    {
        $ids = $message->getData();

        $ids = array_unique(array_filter($ids));
        if (empty($ids)) {
            return;
        }

        $context = $message->getContext();

        RetryableTransaction::retryable($this->connection, function D) use ($ids): void {
            $this->connection->executeStatement(
                'DELETE FROM theme_child WHERE parent_id IN (:ids)',
                ['ids' => Uuid::fromHexToBytesList($ids)],
                ['ids' => ArrayParameterType::STRING]
            );

            $this->connection->executeStatement(
                'INSERT IGNORE INTO theme_child (child_id, parent_id) ( SELECT id as child_id, parent_theme_id as parent_id FROM theme WHERE parent_theme_id IS NOT NULL AND id IN (:ids) AND technical_name IS NULL ) ',

$message->getContext(),
            $criteria
        );

        $binary = Uuid::fromHexToBytes($id);

        $ids = $this->connection->fetchFirstColumn(
            'SELECT LOWER(HEX(product_id)) FROM product_stream_mapping WHERE product_stream_id = :id',
            ['id' => $binary],
        );

        RetryableTransaction::retryable($this->connection, function D) use ($binary): void {
            $this->connection->executeStatement(
                'DELETE FROM product_stream_mapping WHERE product_stream_id = :id',
                ['id' => $binary],
            );
        });

        while ($matches = $iterator->fetchIds()) {
            foreach ($matches as $id) {
                if (!\is_string($id)) {
                    continue;
                }

/** * @param array<string> $parentIds */
    public function update(string $entity, array $parentIds, Context $context): void
    {
        $definition = $this->registry->getByEntityName($entity);

        if (empty($parentIds)) {
            return;
        }

        RetryableTransaction::retryable($this->connection, function D) use ($definition, $parentIds, $context): void {
            $this->trySingleUpdate($definition, $parentIds, $context);
        });
    }

    /** * @param array<string> $parentIds */
    private function trySingleUpdate(EntityDefinition $definition, array $parentIds, Context $context): void
    {
        $entity = $definition->getEntityName();
        $versionAware = $definition->isVersionAware();

if (\array_key_exists('parent', $entity)) {
                $level = $entity['parent']['parentCount'] + 1;
            }

            $query->set($field->getStorageName(), (string) $level);
        }

        $query->andWhere('id = :id');
        $query->setParameter('id', $entity['id']);
        $this->makeQueryVersionAware($definition, Uuid::fromHexToBytes($context->getVersionId()), $query);

        RetryableQuery::retryable($this->connection, function D) use ($query): void {
            $query->executeStatement();
        });

        return Uuid::fromBytesToHex($entity['id']);
    }

    private function buildPathArray(array $parent, TreePathField $field): array
    {
        $path = [];

        if (\array_key_exists('parent', $parent)) {

Profiler::trace('product:indexer:search-keywords', function D) use ($ids, $context): void {
                $this->searchKeywordUpdater->update($ids, $context);
            });
        }

        if ($message->allow(self::STATES_UPDATER)) {
            Profiler::trace('product:indexer:states', function D) use ($ids, $context): void {
                $this->statesUpdater->update($ids, $context);
            });
        }

        RetryableQuery::retryable($this->connection, function D) use ($ids): void {
            $this->connection->executeStatement(
                'UPDATE product SET updated_at = :now WHERE id IN (:ids)',
                ['ids' => Uuid::fromHexToBytesList($ids), 'now' => (new \DateTime())->format(Defaults::STORAGE_DATE_TIME_FORMAT)],
                ['ids' => ArrayParameterType::STRING]
            );
        });

        Profiler::trace('product:indexer:event', function D) use ($ids, $context, $message): void {
            $this->eventDispatcher->dispatch(new ProductIndexerEvent($ids, $context, $message->getSkip()));
        });
    }

if (\count($results) === 0) {
            return [];
        }

        foreach ($results as $row) {
            $affectedIds[] = $row['id'];

            $tags[] = Uuid::fromBytesToHex($row['id']);
        }

        RetryableQuery::retryable($this->connection, function D) use ($affectedIds, $deleteId): void {
            $sqlStatement = ' UPDATE promotion SET promotion.exclusion_ids = JSON_REMOVE(promotion.exclusion_ids, JSON_UNQUOTE(JSON_SEARCH(promotion.exclusion_ids,\'one\', :value))) WHERE id IN(:affectedIds) ';

            $this->connection->executeStatement($sqlStatement, ['value' => $deleteId, 'affectedIds' => $affectedIds], ['affectedIds' => ArrayParameterType::STRING]);
        });

        return $tags;
    }

->andWhere('cluster = :cluster')
            ->setParameter('updatedAt', (new \DateTime())->format(Defaults::STORAGE_DATE_TIME_FORMAT))
            ->setParameter('cluster', $cluster)
            ->setParameter('count', 0)
            ->setParameter('pool', $this->poolName);

        if ($key !== null) {
            $query->andWhere('`key` = :key')
                ->setParameter('key', $key);
        }

        RetryableQuery::retryable($this->connection, function D) use ($query): void {
            $query->executeStatement();
        });
    }

    public function list(string $cluster, int $limit = 5, int $offset = 0): array
    {
        $sql = 'SELECT `key` as array_key, `pool`, `cluster`, `key`, `count` FROM `increment` WHERE `cluster` = :cluster AND `pool` = :pool ORDER BY `count` DESC, `updated_at` DESC';

        $payload = [