emails/api/src/Repository/Job.php

<?php
namespace ProVM\Emails\Repository;

use PDO;
use ProVM\Common\Factory;
use Psr\Log\LoggerInterface;
use ProVM\Common\Define\Model as ModelInterface;
use ProVM\Common\Implement\Repository;
use ProVM\Emails\Model\Job as BaseModel;
use ProVM\Emails;

class Job extends Repository
{
    public function __construct(PDO $connection, LoggerInterface $logger, Factory\Model $factory)
    {
        parent::__construct($connection, $logger);
        $this->setFactory($factory)
            ->setTable('jobs');
    }

    protected Factory\Model $factory;
    public function getFactory(): Factory\Model
    {
        return $this->factory;
    }
    public function setFactory(Factory\Model $factory): Job
    {
        $this->factory = $factory;
        return $this;
    }

    public function install(): void
    {
        $query = "
CREATE TABLE {$this->getTable()} (
    `id` INT UNSIGNED NOT NULL AUTO_INCREMENT,
    `command` VARCHAR(100) NOT NULL,
    `arguments` TEXT NOT NULL,
    PRIMARY KEY (`id`) 
)";
        $this->getConnection()->query($query);
    }

    protected function fieldsForUpdate(): array
    {
        return $this->fieldsForInsert();
    }
    protected function valuesForUpdate(ModelInterface $model): array
    {
        return $this->valuesForInsert($model);
    }
    protected function fieldsForInsert(): array
    {
        return [
            'command',
            'arguments',
        ];
    }
    protected function valuesForInsert(ModelInterface $model): array
    {
        return [
            $model->getCommand(),
            $model->getArguments(),
        ];
    }
    protected function defaultFind(ModelInterface $model): ModelInterface
    {
        return $this->fetchByCommandAndArguments($model->getCommand(), $model->getArguments());
    }
    protected function fieldsForCreate(): array
    {
        return [
            'command',
            'arguments',
        ];
    }
    protected function valuesForCreate(array $data): array
    {
        return [
            $data['command'],
            $data['arguments'],
        ];
    }
    protected function defaultSearch(array $data): ModelInterface
    {
        return $this->fetchByCommandAndArguments($data['command'], $data['arguments']);
    }

    public function load(array $row): ModelInterface
    {
        return (new BaseModel())
            ->setId($row['id'])
            ->setCommand($row['command'])
            ->setArguments($row['arguments'])
            ->setStateRepository($this->getFactory()->find(Emails\Model\State\Job::class));
    }

    public function fetchAllPending(): array
    {
        $query = "SELECT a.*
FROM `{$this->getTable()}` a
    JOIN (SELECT s1.* FROM `jobs_states` s1 JOIN (SELECT MAX(id) AS id, job_id FROM `jobs_states` GROUP BY job_id) s2 ON s2.id = s1.id) b ON b.`job_id` = a.`id` 
WHERE b.`status` = ?";
        return $this->fetchMany($query, [Emails\Model\State\Job::Pending]);
    }
    public function fetchByCommandAndArguments(string $command, string $arguments): Emails\Model\Job
    {
        $query = "SELECT * FROM {$this->getTable()} WHERE `command` = ? AND `arguments` = ?";
        return $this->fetchOne($query, [$command, $arguments]);
    }
    public function fetchAllPendingByCommand(string $command): array
    {
        $query = "SELECT a.* 
FROM `{$this->getTable()}` a
    JOIN (SELECT s1.* FROM `jobs_states` s1 JOIN (SELECT MAX(id) AS id, job_id FROM `jobs_states` GROUP BY job_id) s2 ON s2.id = s1.id) b ON b.`job_id` = a.`id`
WHERE a.`command` = ? AND b.`status` = ?";
        return $this->fetchMany($query, [$command, Emails\Model\State\Job::Pending]);
    }
}