oficial/app/src/Service/Queue.php

<?php
namespace Incoviba\Service;

use Exception;
use Psr\Http\Message\RequestInterface;
use Psr\Log\LoggerInterface;
use Incoviba\Common\Ideal;
use Incoviba\Exception\ServiceAction\{Create, Delete, Read, Update};
use Incoviba\Service;
use Incoviba\Model;

class Queue extends Ideal\Service
{
    public function __construct(LoggerInterface $logger, protected Service\Job $jobService, Worker $defaultWorker,
                                protected int $maxRetries = 5)
    {
        parent::__construct($logger);
        $this->register('default', $defaultWorker);
    }

    protected array $workers;
    public function register(string $name, Worker $worker): self
    {
        $this->workers[strtolower($name)] = $worker;
        return $this;
    }

    public function enqueue(array $configuration): bool
    {
        try {
            $this->jobService->add($configuration);
            return true;
        } catch (Create $exception) {
            $final = new Exception("Could not enqueue job", 0, $exception);
            $this->logger->warning($final);
            return false;
        }
    }
    public function push(array $configuration): bool
    {
        return $this->enqueue($configuration);
    }

    public function runJob(Model\Job $job, ?RequestInterface $request = null): bool
    {
        $type = 'default';
        if (isset($job->configuration['type'])) {
            $type = strtolower($job->configuration['type']);
        }
        if (!isset($this->workers[$type])) {
            $type = 'default';
        }

        $worker = $this->workers[$type];
        if (is_a($worker, Service\Worker\Request::class) and $request !== null) {
            $worker->setRequest($request);
        }

        try {
            if (!$worker->execute($job)) {
                $this->logger->debug("Could not execute job {$job->id}");
                $job->retries++;
                $this->jobService->update($job);
                return false;
            }
            if (!$this->jobService->execute($job)) {
                $this->logger->debug("Could not remove job {$job->id}");
                return false;
            }
        } catch (Exception $exception) {
            $this->logger->warning("Could not run job {$job->id}", ['exception' => [
                'code' => $exception->getCode(),
                'message' => $exception->getMessage(),
                'file' => $exception->getFile(),
                'line' => $exception->getLine(),
                'trace' => $exception->getTraceAsString(),
            ]]);
            $job->retries++;
            try {
                $this->jobService->update($job);
            } catch (Update $exception) {
                $this->logger->error($exception->getMessage(), ['job' => $job, 'exception' => $exception]);
            }
            return false;
        }
        return true;
    }
    public function run(?RequestInterface $request = null): bool
    {
        if (!$this->jobService->isPending()) {
            return true;
        }
        try {
            $job = $this->jobService->get();
        } catch (Read $exception) {
            $this->logger->error($exception->getMessage(), ['exception' => $exception]);
            return false;
        }
        if ($job->retries >= $this->maxRetries) {
            try {
                $this->jobService->remove($job);
            } catch (Delete $exception) {
                $this->logger->error($exception->getMessage(), ['job' => $job, 'exception' => $exception]);
            }
            return true;
        }
        try {
            $this->runJob($job, $request);
        } catch (Exception) {
            $job->retries ++;
            try {
                $this->jobService->update($job);
            } catch (Update $exception) {
                $this->logger->error($exception->getMessage(), ['job' => $job, 'exception' => $exception]);
            }
            return false;
        }
        return true;
    }
}