JobRepository.php

<?php

declare(strict_types=1);

namespace CodeRhapsodie\DataflowBundle\Repository;

use CodeRhapsodie\DataflowBundle\Entity\Job;
use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
use Doctrine\DBAL\Connection;
use Doctrine\DBAL\ParameterType;
use Doctrine\DBAL\Query\QueryBuilder;

/**
 * Repository.
 *
 * @codeCoverageIgnore
 */
class JobRepository
{
    use InitFromDbTrait;

    public const TABLE_NAME = 'cr_dataflow_job';

    /**
     * JobRepository constructor.
     */
    public function __construct(private Connection $connection)
    {
    }

    public function find(int $jobId)
    {
        $qb = $this->createQueryBuilder();
        $qb
            ->andWhere($qb->expr()->eq('id', $qb->createNamedParameter($jobId, ParameterType::INTEGER)))
        ;

        return $this->returnFirstOrNull($qb);
    }

    public function findOneshotDataflows(): iterable
    {
        $qb = $this->createQueryBuilder();
        $qb
            ->andWhere($qb->expr()->isNull('scheduled_dataflow_id'))
            ->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, ParameterType::INTEGER)));
        $stmt = $qb->executeQuery();
        if ($stmt->rowCount() === 0) {
            return [];
        }
        while (false !== ($row = $stmt->fetchAssociative())) {
            yield Job::createFromArray($this->initDateTime($this->initArray($row)));
        }
    }

    public function findPendingForScheduledDataflow(ScheduledDataflow $scheduled): ?Job
    {
        $qb = $this->createQueryBuilder();
        $qb
            ->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($scheduled->getId(), ParameterType::INTEGER)))
            ->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, ParameterType::INTEGER)));

        return $this->returnFirstOrNull($qb);
    }

    public function findNextPendingDataflow(): ?Job
    {
        $qb = $this->createQueryBuilder();
        $qb->andWhere($qb->expr()->lte('requested_date', $qb->createNamedParameter(new \DateTime(), 'datetime')))
            ->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, ParameterType::INTEGER)))
            ->orderBy('requested_date', 'ASC')
            ->setMaxResults(1)
        ;

        return $this->returnFirstOrNull($qb);
    }

    public function findLastForDataflowId(int $dataflowId): ?Job
    {
        $qb = $this->createQueryBuilder();
        $qb->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($dataflowId, ParameterType::INTEGER)))
            ->orderBy('requested_date', 'DESC')
            ->setMaxResults(1)
        ;

        return $this->returnFirstOrNull($qb);
    }

    public function findLatests(): iterable
    {
        $qb = $this->createQueryBuilder();
        $qb
            ->orderBy('requested_date', 'DESC')
            ->setMaxResults(20);
        $stmt = $qb->executeQuery();
        if ($stmt->rowCount() === 0) {
            return [];
        }
        while (false !== ($row = $stmt->fetchAssociative())) {
            yield Job::createFromArray($row);
        }
    }

    public function findForScheduled(int $id): iterable
    {
        $qb = $this->createQueryBuilder();
        $qb->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($id, ParameterType::INTEGER)))
            ->orderBy('requested_date', 'DESC')
            ->setMaxResults(20);
        $stmt = $qb->executeQuery();
        if ($stmt->rowCount() === 0) {
            return [];
        }
        while (false !== ($row = $stmt->fetchAssociative())) {
            yield Job::createFromArray($row);
        }
    }

    public function save(Job $job)
    {
        $datas = $job->toArray();
        unset($datas['id']);

        if (\is_array($datas['options'])) {
            $datas['options'] = json_encode($datas['options'], \JSON_THROW_ON_ERROR);
        }
        if (\is_array($datas['exceptions'])) {
            $datas['exceptions'] = json_encode($datas['exceptions'], \JSON_THROW_ON_ERROR);
        }

        if ($job->getId() === null) {
            $this->connection->insert(static::TABLE_NAME, $datas, $this->getFields());
            $job->setId((int) $this->connection->lastInsertId());

            return;
        }
        $this->connection->update(static::TABLE_NAME, $datas, ['id' => $job->getId()], $this->getFields());
    }

    public function updateCount(int $jobId, int $count): void
    {
        $this->connection->update(static::TABLE_NAME, ['count' => $count], ['id' => $jobId]);
    }

    public function createQueryBuilder($alias = null): QueryBuilder
    {
        $qb = $this->connection->createQueryBuilder();
        $qb->select('*')
            ->from(static::TABLE_NAME, $alias);

        return $qb;
    }

    public function crashLongRunning(int $hours): void
    {
        $qb = $this->connection->createQueryBuilder();
        $qb->update(static::TABLE_NAME, 'j')
            ->set('j.status', ':new_status')
            ->set('j.end_time', ':now')
            ->andWhere('j.status = :status')
            ->andWhere('j.start_time < :date')
            ->setParameter('status', Job::STATUS_RUNNING)
            ->setParameter('date', new \DateTime("- {$hours} hours"), 'datetime')
            ->setParameter('new_status', Job::STATUS_CRASHED)
            ->setParameter('now', new \DateTime(), 'datetime')
            ->executeStatement()
        ;
    }

    public function deleteOld(int $days): void
    {
        $qb = $this->connection->createQueryBuilder();
        $qb->delete(static::TABLE_NAME, 'j')
            ->andWhere($qb->expr()->in('j.status', [Job::STATUS_COMPLETED, Job::STATUS_CRASHED]))
            ->andWhere('j.end_time < :date')
            ->setParameter('date', new \DateTime("- {$days} days"), 'datetime')
            ->executeStatement()
        ;
    }

    private function returnFirstOrNull(QueryBuilder $qb): ?Job
    {
        $stmt = $qb->executeQuery();
        if ($stmt->rowCount() === 0) {
            return null;
        }

        return Job::createFromArray($this->initDateTime($this->initArray($stmt->fetchAssociative())));
    }

    private function getFields(): array
    {
        return [
            'id' => ParameterType::INTEGER,
            'status' => ParameterType::INTEGER,
            'label' => ParameterType::STRING,
            'dataflow_type' => ParameterType::STRING,
            'options' => ParameterType::STRING,
            'requested_date' => 'datetime',
            'scheduled_dataflow_id' => ParameterType::INTEGER,
            'count' => ParameterType::INTEGER,
            'exceptions' => ParameterType::STRING,
            'start_time' => 'datetime',
            'end_time' => 'datetime',
        ];
    }
}