esun3d/app/code/Esun/MultiLocaleTranslation/Model/Queue/QueueProcessor.php

<?php

declare(strict_types=1);

namespace Esun\MultiLocaleTranslation\Model\Queue;

use Esun\MultiLocaleTranslation\Model\EntityPool;
use Esun\MultiLocaleTranslation\Model\LocaleResolver;
use Esun\MultiLocaleTranslation\Model\ProviderPool;
use Esun\MultiLocaleTranslation\Model\TranslationLogger;
use Magento\Framework\App\ResourceConnection;
use Magento\Framework\DB\Sql\Expression;

class QueueProcessor
{
    private const TABLE_NAME = 'esun_localetranslation_queue';

    public function __construct(
        private readonly ResourceConnection $resourceConnection,
        private readonly EntityPool $entityPool,
        private readonly ProviderPool $providerPool,
        private readonly LocaleResolver $localeResolver,
        private readonly TranslationLogger $translationLogger
    ) {
    }

    public function process(int $limit = 10, int $timeLimit = 50): int
    {
        $processed = 0;
        $startedAt = microtime(true);
        $limit = min(50, max(1, $limit));
        $timeLimit = min(300, max(10, $timeLimit));

        for ($index = 0; $index < $limit; $index++) {
            if ($processed > 0 && microtime(true) - $startedAt >= $timeLimit) {
                break;
            }
            $task = $this->claimNext();
            if ($task === null) {
                break;
            }
            $this->processTask($task);
            $processed++;
        }

        return $processed;
    }

    /** @return array<string, mixed>|null */
    private function claimNext(): ?array
    {
        $connection = $this->resourceConnection->getConnection();
        $table = $this->resourceConnection->getTableName(self::TABLE_NAME);

        while (true) {
            $queueId = (int)$connection->fetchOne(
                $connection->select()->from($table, 'queue_id')
                    ->where('status = ?', 'pending')
                    ->order('queue_id ASC')
                    ->limit(1)
            );
            if ($queueId <= 0) {
                return null;
            }

            $affected = $connection->update(
                $table,
                [
                    'status' => 'processing',
                    'attempts' => new Expression('attempts + 1'),
                    'started_at' => new Expression('CURRENT_TIMESTAMP'),
                    'finished_at' => null,
                    'error_message' => null,
                ],
                ['queue_id = ?' => $queueId, 'status = ?' => 'pending']
            );
            if ($affected === 1) {
                $task = $connection->fetchRow(
                    $connection->select()->from($table)->where('queue_id = ?', $queueId)
                );
                return is_array($task) ? $task : null;
            }
        }
    }

    /** @param array<string, mixed> $task */
    private function processTask(array $task): void
    {
        $queueId = (int)$task['queue_id'];
        $provider = (string)$task['provider'];
        $entityType = (string)$task['entity_type'];
        $entityId = (int)$task['entity_id'];
        $sourceStoreId = (int)$task['source_store_id'];
        $targetStoreId = (int)$task['target_store_id'];
        $sourceLanguage = '';
        $targetLanguage = '';
        $providerFields = [];

        try {
            $entity = $this->entityPool->get($entityType);
            foreach ($entity->getFields($entityId, $sourceStoreId, $targetStoreId) as $field) {
                $value = (string)($field['source_value'] ?? '');
                if ($value === '' || empty($field['code'])) {
                    continue;
                }
                $providerFields[(string)$field['code']] = [
                    'value' => $value,
                    'is_html' => !empty($field['is_html']),
                ];
            }

            $sourceLanguage = $this->localeResolver->getLanguageCode($sourceStoreId, $provider, false);
            $targetLanguage = $this->localeResolver->getLanguageCode($targetStoreId, $provider, true);
            if ($providerFields === []) {
                $translations = [];
            } elseif ($sourceLanguage !== '' && $sourceLanguage === $targetLanguage) {
                $translations = $this->copyValues($providerFields);
            } else {
                $translations = $this->providerPool->get($provider)->translate(
                    $providerFields,
                    $sourceLanguage,
                    $targetLanguage
                );
            }

            $entity->apply($entityId, $targetStoreId, $translations);
            $this->safeLogSuccess(
                $provider,
                $entityType,
                $entityId,
                $sourceStoreId,
                $targetStoreId,
                $sourceLanguage,
                $targetLanguage,
                $providerFields,
                $translations
            );
            $this->finish($queueId, 'success', null);
        } catch (\Throwable $exception) {
            $this->safeLogFailure(
                $provider,
                $entityType,
                $entityId,
                $sourceStoreId,
                $targetStoreId,
                $sourceLanguage,
                $targetLanguage,
                $providerFields,
                $exception->getMessage()
            );
            $this->finish($queueId, 'failed', $exception->getMessage());
        }
    }

    /** @param array<string, mixed> $fields @return array<string, string> */
    private function copyValues(array $fields): array
    {
        $values = [];
        foreach ($fields as $code => $field) {
            $values[$code] = (string)($field['value'] ?? '');
        }
        return $values;
    }

    private function finish(int $queueId, string $status, ?string $error): void
    {
        $connection = $this->resourceConnection->getConnection();
        $connection->update(
            $this->resourceConnection->getTableName(self::TABLE_NAME),
            [
                'status' => $status,
                'error_message' => $error,
                'finished_at' => new Expression('CURRENT_TIMESTAMP'),
            ],
            ['queue_id = ?' => $queueId]
        );
    }

    private function safeLogSuccess(
        string $provider,
        string $entityType,
        int $entityId,
        int $sourceStoreId,
        int $targetStoreId,
        string $sourceLanguage,
        string $targetLanguage,
        array $fields,
        array $translations
    ): void {
        try {
            $this->translationLogger->logSuccess(
                $provider,
                $entityType,
                $entityId,
                $sourceStoreId,
                $targetStoreId,
                $sourceLanguage,
                $targetLanguage,
                $fields,
                $translations,
                $sourceLanguage === $targetLanguage ? 'copied' : 'success'
            );
        } catch (\Throwable) {
        }
    }

    private function safeLogFailure(
        string $provider,
        string $entityType,
        int $entityId,
        int $sourceStoreId,
        int $targetStoreId,
        string $sourceLanguage,
        string $targetLanguage,
        array $fields,
        string $error
    ): void {
        try {
            $this->translationLogger->logFailure(
                $provider,
                $entityType,
                $entityId,
                $sourceStoreId,
                $targetStoreId,
                $sourceLanguage,
                $targetLanguage,
                $fields,
                $error
            );
        } catch (\Throwable) {
        }
    }
}