<?php
declare(strict_types=1);

/**
 * BEdita, API-first content management framework
 * Copyright 2024 ChannelWeb Srl, Chialab Srl
 *
 * This file is part of BEdita: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published
 * by the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * See LICENSE.LGPL or <http://gnu.org/licenses/lgpl-3.0.html> for more details.
 */
namespace BEdita\Core\Model\Table;

use BEdita\Core\Job\QueueJob;
use BEdita\Core\Model\Entity\AsyncJob;
use BEdita\Core\Model\Validation\Validation;
use Cake\Core\Configure;
use Cake\Database\Expression\QueryExpression;
use Cake\Database\Schema\TableSchemaInterface;
use Cake\Datasource\ConnectionManager;
use Cake\Event\EventInterface;
use Cake\I18n\FrozenTime;
use Cake\ORM\Query;
use Cake\ORM\Table;
use Cake\Queue\QueueManager;
use Cake\Validation\Validator;

/**
 * AsyncJobs Model
 *
 * @method \BEdita\Core\Model\Entity\AsyncJob get($primaryKey, $options = [])
 * @method \BEdita\Core\Model\Entity\AsyncJob newEntity($data = null, array $options = [])
 * @method \BEdita\Core\Model\Entity\AsyncJob[] newEntities(array $data, array $options = [])
 * @method \BEdita\Core\Model\Entity\AsyncJob|bool save(\Cake\Datasource\EntityInterface $entity, $options = [])
 * @method \BEdita\Core\Model\Entity\AsyncJob patchEntity(\Cake\Datasource\EntityInterface $entity, array $data, array $options = [])
 * @method \BEdita\Core\Model\Entity\AsyncJob[] patchEntities($entities, array $data, array $options = [])
 * @method \BEdita\Core\Model\Entity\AsyncJob findOrCreate($search, callable $callback = null, $options = [])
 * @mixin \Cake\ORM\Behavior\TimestampBehavior
 * @since 4.0.0
 */
class AsyncJobsTable extends Table
{
    /**
     * @inheritDoc
     */
    public static function defaultConnectionName(): string
    {
        if (in_array('async_jobs', ConnectionManager::configured())) {
            return 'async_jobs';
        }

        return parent::defaultConnectionName();
    }

    /**
     * {@inheritDoc}
     *
     * @codeCoverageIgnore
     */
    public function initialize(array $config): void
    {
        parent::initialize($config);

        $this->setTable('async_jobs');

        $this->setPrimaryKey('uuid');
        $this->setDisplayField('payload');

        $this->addBehavior('Timestamp', [
            'events' => [
                'Model.beforeSave' => [
                    'created' => 'new',
                    'modified' => 'always',
                ],
                'AsyncJob.complete' => [
                    'completed' => 'always',
                ],
            ],
        ]);
    }

    /**
     * {@inheritDoc}
     *
     * @codeCoverageIgnore
     */
    public function validationDefault(Validator $validator): Validator
    {
        return $validator
            ->uuid('uuid')
            ->allowEmptyString('uuid', null, 'create')

            ->requirePresence('service', 'create')
            ->notEmptyString('service')

            ->naturalNumber('priority')
            ->allowEmptyString('priority')

            ->allowEmptyString('payload')

            ->add('scheduled_from', 'dateTime', ['rule' => [Validation::class, 'dateTime']])
            ->allowEmptyDateTime('scheduled_from')

            ->add('expires', 'dateTime', ['rule' => [Validation::class, 'dateTime']])
            ->allowEmptyDateTime('expires')

            ->naturalNumber('max_attempts')
            ->notEmptyString('max_attempts')

            ->add('locked_until', 'dateTime', ['rule' => [Validation::class, 'dateTime']])
            ->allowEmptyDateTime('locked_until')

            ->add('completed', 'dateTime', ['rule' => [Validation::class, 'dateTime']])
            ->allowEmptyDateTime('completed')

            ->allowEmptyString('results');
    }

    /**
     * {@inheritDoc}
     *
     * @codeCoverageIgnore
     */
    public function getSchema(): TableSchemaInterface
    {
        return parent::getSchema()
            ->setColumnType('payload', 'json')
            ->setColumnType('uuid', 'uuid')
            ->setColumnType('results', 'json');
    }

    /**
     * Queue async job as new entity is created.
     *
     * @param \Cake\Event\EventInterface $event The event
     * @param \BEdita\Core\Model\Entity\AsyncJob $entity The entity persisted
     * @return void
     */
    public function afterSave(EventInterface $event, AsyncJob $entity): void
    {
        if (!$entity->isNew() || !QueueManager::getConfig('default')) {
            return;
        }

        $delay = (int)Configure::read('Queue.default.pushDelay', 1);
        if ($entity->scheduled_from && $entity->scheduled_from->isFuture()) {
            $delay = $entity->scheduled_from->diffInSeconds();
        }

        $expires = null;
        if ($entity->expires && $entity->expires->isFuture()) {
            $expires = $entity->expires->diffInSeconds();
        }

        QueueManager::push(
            QueueJob::class,
            ['uuid' => $entity->uuid],
            compact('delay', 'expires'),
        );
    }

    /**
     * Lock an asynchronous job for execution.
     *
     * @param string $uuid UUID of job to be locked.
     * @param mixed $duration Duration. By default, jobs are locked for 5 minutes.
     * @return \BEdita\Core\Model\Entity\AsyncJob
     */
    public function lock($uuid, $duration = '+5 minutes')
    {
        return $this->getConnection()->transactional(function () use ($uuid, $duration) {
            $entity = $this->get($uuid, ['finder' => 'pending']);
            $entity->max_attempts -= 1;
            $entity->locked_until = new FrozenTime($duration);

            $expires = $entity->locked_until->timestamp;
            $this->dispatchEvent('AsyncJob.lock', compact('entity', 'expires'));

            return $this->saveOrFail($entity, ['atomic' => false]);
        });
    }

    /**
     * Unlock an asynchronous job after execution (either successful or failed).
     *
     * @param string $uuid UUID of job to be unlocked.
     * @param mixed $success Job run success. If strictly equal to `false`, job is considered failed.
     * @return void
     */
    public function unlock($uuid, $success)
    {
        $this->getConnection()->transactional(function () use ($uuid, $success) {
            $entity = $this->get($uuid);
            $entity->locked_until = null;

            $event = 'AsyncJob.fail';
            if ($success !== false) {
                $event = 'AsyncJob.complete';
            }
            $this->dispatchEvent($event, compact('entity', 'success'));

            $this->saveOrFail($entity, ['atomic' => false]);
        });
    }

    /**
     * Finder for pending jobs.
     *
     * This finder returns a query object that filters asynchronous jobs that are
     * still valid (not completed, not yet expired, not locked, and have some attempts left).
     *
     * @param \Cake\ORM\Query $query Query object instance.
     * @return \Cake\ORM\Query
     */
    protected function findPending(Query $query)
    {
        $now = FrozenTime::now();

        return $query->where(fn (QueryExpression $exp): QueryExpression => $exp->and([
            $exp->or(
                fn (QueryExpression $exp): QueryExpression => $exp
                    ->isNull($this->aliasField('scheduled_from'))
                    ->lte($this->aliasField('scheduled_from'), $now),
            ),
            $exp->or(
                fn (QueryExpression $exp): QueryExpression => $exp
                    ->isNull($this->aliasField('expires'))
                    ->gte($this->aliasField('expires'), $now),
            ),
            $exp->or(
                fn (QueryExpression $exp): QueryExpression => $exp
                    ->isNull($this->aliasField('locked_until'))
                    ->lt($this->aliasField('locked_until'), $now),
            ),
            fn (QueryExpression $exp): QueryExpression => $exp
                ->gt($this->aliasField('max_attempts'), 0)
                ->isNull($this->aliasField('completed')),
        ]));
    }

    /**
     * Finder for failed async jobs.
     *
     * This finder returns a query object that filters asynchronous jobs that are either expired or
     * that have failed too many times.
     *
     * @param \Cake\ORM\Query $query Query object instance.
     * @return \Cake\ORM\Query
     */
    protected function findFailed(Query $query)
    {
        $now = FrozenTime::now();

        return $query->where(fn (QueryExpression $exp): QueryExpression => $exp->and([
            fn (QueryExpression $exp): QueryExpression => $exp->isNull($this->aliasField('completed')),
            $exp->or([
                fn (QueryExpression $exp): QueryExpression => $exp->lt($this->aliasField('expires'), $now),
                $exp->and([
                    fn (QueryExpression $exp): QueryExpression => $exp->eq($this->aliasField('max_attempts'), 0),
                    $exp->or(
                        fn (QueryExpression $exp): QueryExpression => $exp
                            ->isNull($this->aliasField('locked_until'))
                            ->lt($this->aliasField('locked_until'), $now),
                    ),
                ]),
            ]),
        ]));
    }

    /**
     * Finder for completed async jobs.
     *
     * This finder returns a query object that filters asynchronous jobs that have been completed successfully.
     *
     * @param \Cake\ORM\Query $query Query object instance.
     * @return \Cake\ORM\Query
     */
    protected function findCompleted(Query $query)
    {
        return $query->where(
            fn (QueryExpression $exp): QueryExpression => $exp->isNotNull($this->aliasField('completed')),
        );
    }

    /**
     * Finder for incomplete async jobs.
     *
     * This finder returns a query object that filters asynchronous jobs that haven't been completed yet.
     *
     * @param \Cake\ORM\Query $query Query object instance.
     * @return \Cake\ORM\Query
     */
    protected function findIncomplete(Query $query)
    {
        return $query->where(
            fn (QueryExpression $exp): QueryExpression => $exp->isNull($this->aliasField('completed')),
        );
    }

    /**
     * Find pending asynchronous jobs sorted by descending priority, and optionally filtered by service and priority.
     *
     * @param \Cake\ORM\Query $query Query object instance.
     * @param array $options Additional options.
     * @return \Cake\ORM\Query
     */
    protected function findPriority(Query $query, array $options)
    {
        $options = array_filter(array_intersect_key($options, array_flip(['priority', 'service'])));
        if (!empty($options)) {
            $query = $query->where(function (QueryExpression $exp) use ($options) {
                if (!empty($options['priority'])) {
                    $exp->gte($this->aliasField('priority'), $options['priority']);
                }
                if (!empty($options['service'])) {
                    $exp->eq($this->aliasField('service'), $options['service']);
                }

                return $exp;
            });
        }

        return $query
            ->find('pending')
            ->orderDesc($this->aliasField('priority'));
    }

    /**
     * Update field "results" by entity, success and message
     *
     * @param \BEdita\Core\Model\Entity\AsyncJob $entity The Job entity
     * @param bool $success The success flag
     * @param array $messages The messages
     * @return void
     */
    public function updateResults(AsyncJob $entity, bool $success, array $messages = []): void
    {
        $this->getConnection()->transactional(function () use ($entity, $success, $messages) {
            $results = (array)$entity->get('results');
            $attempt = count($results) + 1;
            $data = compact('messages');
            $result = compact('data', 'success');
            $result['attempt_number'] = $attempt;
            $results[] = $result;
            $entity->set('results', $results);
            $this->saveOrFail($entity, ['atomic' => false]);
        });
    }
}
