QueuedJobsTable.php

<?php
declare(strict_types=1);

namespace Queue\Model\Table;

use ArrayObject;
use Cake\Core\Configure;
use Cake\Core\Plugin;
use Cake\Event\Event;
use Cake\Event\EventInterface;
use Cake\Event\EventManager;
use Cake\I18n\DateTime;
use Cake\Log\Log;
use Cake\ORM\Query\SelectQuery;
use Cake\ORM\Table;
use Cake\Validation\Validator;
use InvalidArgumentException;
use PhpCollective\Dto\Dto\FromArrayToArrayInterface;
use Queue\Config\JobConfig;
use Queue\Model\Entity\QueuedJob;
use Queue\Model\Filter\QueuedJobsCollection;
use Queue\Queue\Config;
use Queue\Queue\TaskFinder;
use Queue\Utility\Memory;

/**
 * @author MGriesbach@gmail.com
 * @license http://www.opensource.org/licenses/mit-license.php The MIT License
 * @extends \Cake\ORM\Table<array{Search: \Search\Model\Behavior\SearchBehavior, Timestamp: \Cake\ORM\Behavior\TimestampBehavior}>
 * @property \Queue\Model\Table\QueueProcessesTable&\Cake\ORM\Association\BelongsTo $WorkerProcesses
 * @method \Queue\Model\Entity\QueuedJob get(mixed $primaryKey, array|string $finder = 'all', \Psr\SimpleCache\CacheInterface|string|null $cache = null, \Closure|string|null $cacheKey = null, mixed ...$args)
 * @method \Queue\Model\Entity\QueuedJob newEntity(array $data, array $options = [])
 * @method array<\Queue\Model\Entity\QueuedJob> newEntities(array $data, array $options = [])
 * @method \Queue\Model\Entity\QueuedJob|false save(\Cake\Datasource\EntityInterface $entity, array $options = [])
 * @method \Queue\Model\Entity\QueuedJob patchEntity(\Cake\Datasource\EntityInterface $entity, array $data, array $options = [])
 * @method array<\Queue\Model\Entity\QueuedJob> patchEntities(iterable $entities, array $data, array $options = [])
 * @method \Queue\Model\Entity\QueuedJob findOrCreate(\Cake\ORM\Query\SelectQuery|callable|array $search, ?callable $callback = null, array $options = [])
 * @method \Queue\Model\Entity\QueuedJob saveOrFail(\Cake\Datasource\EntityInterface $entity, array $options = [])
 * @method \Queue\Model\Entity\QueuedJob newEmptyEntity()
 * @method \Cake\Datasource\ResultSetInterface<\Queue\Model\Entity\QueuedJob>|false saveMany(iterable $entities, array $options = [])
 * @method \Cake\Datasource\ResultSetInterface<\Queue\Model\Entity\QueuedJob> saveManyOrFail(iterable $entities, array $options = [])
 * @method \Cake\Datasource\ResultSetInterface<\Queue\Model\Entity\QueuedJob>|false deleteMany(iterable $entities, array $options = [])
 * @method \Cake\Datasource\ResultSetInterface<\Queue\Model\Entity\QueuedJob> deleteManyOrFail(iterable $entities, array $options = [])
 * @mixin \Cake\ORM\Behavior\TimestampBehavior
 * @mixin \Search\Model\Behavior\SearchBehavior
 */
class QueuedJobsTable extends Table {

	/**
	 * @var string
	 */
	public const DRIVER_MYSQL = 'Mysql';

	/**
	 * @var string
	 */
	public const DRIVER_POSTGRES = 'Postgres';

	/**
	 * @var string
	 */
	public const DRIVER_SQLSERVER = 'Sqlserver';

	/**
	 * @var string
	 */
	public const DRIVER_SQLITE = 'Sqlite';

	/**
	 * @var int
	 */
	public const STATS_LIMIT = 100000;

	/**
	 * @var int
	 */
	public const DAY = 86400;

	/**
	 * Terminal `status` value stamped on a job that has exhausted all of its
	 * retries. Distinguishes a permanently-failed job (which will never run
	 * again) from one that is pending, in progress, or still being retried —
	 * all of which otherwise share `completed IS NULL`.
	 *
	 * @var string
	 */
	public const STATUS_ABORTED = 'aborted';

	/**
	 * @var array<string, string>
	 */
	public array $rateHistory = [];

	/**
	 * @var \Queue\Queue\TaskFinder|null
	 */
	protected ?TaskFinder $taskFinder = null;

	/**
	 * @var string|null
	 */
	protected ?string $_key = null;

	/**
	 * set connection name
	 *
	 * @return string
	 */
	public static function defaultConnectionName(): string {
		$connection = Configure::read('Queue.connection');
		if (!empty($connection)) {
			return $connection;
		}

		return parent::defaultConnectionName();
	}

	/**
	 * initialize Table
	 *
	 * @param array<string, mixed> $config Configuration
	 *
	 * @return void
	 */
	public function initialize(array $config): void {
		parent::initialize($config);

		$this->addBehavior('Timestamp');
		if (Configure::read('Queue.isSearchEnabled') !== false && Plugin::isLoaded('Search')) {
			$this->addBehavior('Search.Search', [
				'collectionClass' => QueuedJobsCollection::class,
			]);
		}

		$this->belongsTo('WorkerProcesses', [
			'className' => 'Queue.QueueProcesses',
			'foreignKey' => false,
			'conditions' => [
				'WorkerProcesses.workerkey = QueuedJobs.workerkey',
			],
		]);

		$this->getSchema()->setColumnType('data', 'json');
	}

	/**
	 * @param \Cake\Event\EventInterface $event
	 * @param \ArrayObject<string, mixed> $data
	 * @param \ArrayObject<string, mixed> $options
	 *
	 * @return void
	 */
	public function beforeMarshal(EventInterface $event, ArrayObject $data, ArrayObject $options): void {
		if (isset($data['data']) && $data['data'] === '') {
			$data['data'] = null;
		}
	}

	/**
	 * Default validation rules.
	 *
	 * @param \Cake\Validation\Validator $validator Validator instance.
	 *
	 * @return \Cake\Validation\Validator
	 */
	public function validationDefault(Validator $validator): Validator {
		$validator
			->integer('id')
			->allowEmptyString('id', null, 'create');

		$validator
			->requirePresence('job_task', 'create')
			->notEmptyString('job_task');

		$validator
			->greaterThanOrEqual('progress', 0)
			->lessThanOrEqual('progress', 1)
			->allowEmptyString('progress');

		return $validator;
	}

	/**
	 * @return \Queue\Config\JobConfig
	 */
	public function createConfig(): JobConfig {
		return new JobConfig();
	}

	/**
	 * Adds a new job to the queue.
	 *
	 * Config
	 * - priority: 1-10, defaults to 5
	 * - notBefore: Optional date which must not be preceded
	 * - group: Used to group similar QueuedJobs
	 * - reference: An optional reference string
	 * - status: To set an initial status text
	 * - unique: When true (with a `reference`), an existing pending job
	 *   matching `(reference, resolved job_task)` is returned instead of
	 *   inserting a duplicate row. Useful for fan-out dispatchers that
	 *   could otherwise stack up identical per-tenant work when a previous
	 *   run is still pending or stuck.
	 *
	 * @param string $jobTask Job task name or FQCN.
	 * @param object|array<string, mixed>|null $data Array of data or DTO like object.
	 * @param \Queue\Config\JobConfig|array<string, mixed> $config Config to save along with the job.
	 *
	 * @throws \InvalidArgumentException If `unique` is set without a `reference`.
	 *
	 * @return \Queue\Model\Entity\QueuedJob Saved job entity (or the existing pending entity if `unique` deduped).
	 */
	public function createJob(string $jobTask, object|array|null $data = null, JobConfig|array $config = []): QueuedJob {
		if (!$config instanceof JobConfig) {
			$config = $this->createConfig()->fromArray($config);
		}

		if (interface_exists(FromArrayToArrayInterface::class) && $data instanceof FromArrayToArrayInterface) {
			$data = $data->toArray();
		} elseif (is_object($data) && method_exists($data, 'toArray')) {
			$data = $data->toArray();
		}
		if ($data !== null && !is_array($data)) {
			throw new InvalidArgumentException('Data must be `array|null`, implement `' . FromArrayToArrayInterface::class . '` or provide a `toArray()` method');
		}

		$resolvedTask = $this->jobTask($jobTask);

		if ($config->isUnique()) {
			if (!$config->hasReference()) {
				throw new InvalidArgumentException('createJob() with `unique` requires a `reference` to dedupe on.');
			}

			/** @var \Queue\Model\Entity\QueuedJob|null $existing */
			$existing = $this->find()
				->where([
					'reference' => $config->getReferenceOrFail(),
					'job_task' => $resolvedTask,
					'completed IS' => null,
				])
				->first();
			if ($existing !== null) {
				Log::info(sprintf(
					'Queue.createJob: dedup hit for %s reference=%s (returning existing job_id=%d)',
					$resolvedTask,
					$config->getReferenceOrFail(),
					$existing->id,
				));

				return $existing;
			}
		}

		$queuedJob = [
			'job_task' => $resolvedTask,
			'data' => $data,
			'notbefore' => $config->hasNotBefore() ? $this->getDateTime($config->getNotBeforeOrFail()) : null,
			'priority' => $config->getPriority(),
		] + $config->toArray();
		if ($queuedJob['priority'] === null) {
			unset($queuedJob['priority']);
		}

		$queuedJob = $this->newEntity($queuedJob);
		$queuedJob = $this->saveOrFail($queuedJob);

		$event = new Event('Queue.Job.created', $this, [
			'job' => $queuedJob,
		]);
		EventManager::instance()->dispatch($event);

		return $queuedJob;
	}

	/**
	 * @param class-string<\Queue\Queue\Task>|string $jobType
	 *
	 * @return string
	 */
	protected function jobTask(string $jobType): string {
		if (!($this->taskFinder instanceof TaskFinder)) {
			$this->taskFinder = new TaskFinder();
		}

		return $this->taskFinder->resolve($jobType);
	}

	/**
	 * @param string $reference
	 * @param string|null $jobTask
	 *
	 * @throws \InvalidArgumentException
	 *
	 * @return bool
	 */
	public function isQueued(string $reference, ?string $jobTask = null): bool {
		if (!$reference) {
			throw new InvalidArgumentException('A reference is needed');
		}

		$conditions = [
			'reference' => $reference,
			'completed IS' => null,
			// A job that exhausted its retries (status = aborted) will never
			// run again, so it is not "queued" anymore even though it has no
			// completed timestamp. Excluding it stops callers that gate on
			// isQueued() — e.g. a non-concurrent scheduler row — from wedging
			// permanently behind a dead job. The OR keeps null/other statuses.
			'OR' => [
				'status IS' => null,
				'status !=' => static::STATUS_ABORTED,
			],
		];
		if ($jobTask) {
			$conditions['job_task'] = $jobTask;
		}

		return (bool)$this->find()->where($conditions)->select(['id'])->first();
	}

	/**
	 * Returns the number of items in the queue.
	 * Either returns the number of ALL pending jobs, or the number of pending jobs of the passed type.
	 *
	 * @param string|null $type Job type to Count
	 *
	 * @return int
	 */
	public function getLength(?string $type = null): int {
		$conditions = $this->pendingConditions();
		if ($type !== null) {
			$conditions['job_task'] = $type;
		}

		return $this->find('all', conditions: $conditions)->count();
	}

	/**
	 * Return a list of all job types in the Queue.
	 *
	 * @return \Cake\ORM\Query\SelectQuery
	 */
	public function getTypes(): SelectQuery {
		return $this->find('list', keyField: 'job_task', valueField: 'job_task')
			->select(['job_task'])
			->groupBy(['job_task']);
	}

	/**
	 * Return some statistics about finished jobs still in the Database.
	 * TO-DO: rewrite as virtual field
	 *
	 * @param bool $disableHydration
	 *
	 * @return array<\Queue\Model\Entity\QueuedJob>|array<mixed>
	 */
	public function getStats(bool $disableHydration = false): array {
		$driverName = $this->getDriverName();
		$options = [
			'fields' => function (SelectQuery $query) use ($driverName) {
				$alltime = $query->func()->avg('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(created)');
				$runtime = $query->func()->avg('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(fetched)');
				$fetchdelay = $query->func()->avg('UNIX_TIMESTAMP(fetched) - IF(notbefore is NULL, UNIX_TIMESTAMP(created), UNIX_TIMESTAMP(notbefore))');
				switch ($driverName) {
					case static::DRIVER_SQLSERVER:
						$alltime = $query->func()->avg("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', created)");
						$runtime = $query->func()->avg("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', fetched)");
						$fetchdelay = $query->func()->avg("DATEDIFF(s, '1970-01-01 00:00:00', fetched) - (CASE WHEN notbefore IS NULL THEN DATEDIFF(s, '1970-01-01 00:00:00', created) ELSE DATEDIFF(s, '1970-01-01 00:00:00', notbefore) END)");

						break;
					case static::DRIVER_POSTGRES:
						$alltime = $query->func()->avg('EXTRACT(EPOCH FROM completed) - EXTRACT(EPOCH FROM created)');
						$runtime = $query->func()->avg('EXTRACT(EPOCH FROM completed) - EXTRACT(EPOCH FROM fetched)');
						$fetchdelay = $query->func()->avg('EXTRACT(EPOCH FROM fetched) - CASE WHEN notbefore IS NULL then EXTRACT(EPOCH FROM created) ELSE EXTRACT(EPOCH FROM notbefore) END');

						break;
					case static::DRIVER_SQLITE:
						$alltime = $query->func()->avg('julianday(completed) - julianday(created)');
						$runtime = $query->func()->avg('julianday(completed) - julianday(fetched)');
						$fetchdelay = $query->func()->avg('julianday(fetched) - (CASE WHEN notbefore IS NULL THEN julianday(created) ELSE julianday(notbefore) END)');

						break;
				}

				return [
					'job_task',
					'num' => $query->func()->count('*'),
					'alltime' => $alltime,
					'runtime' => $runtime,
					'fetchdelay' => $fetchdelay,
				];
			},
			'conditions' => [
				'completed IS NOT' => null,
			],
		];

		$query = $this->find('all', ...$options)
			->groupBy(['job_task']);
		if ($disableHydration) {
			$query = $query->disableHydration();
		}
		$result = $query->toArray();
		if ($result && $driverName === static::DRIVER_SQLITE) {
			foreach ($result as $key => $row) {
				$result[$key]['fetchdelay'] = (int)round($row['fetchdelay'] * static::DAY);
				$result[$key]['runtime'] = (int)round($row['runtime'] * static::DAY);
				$result[$key]['alltime'] = (int)round($row['alltime'] * static::DAY);
			}
		}

		return $result;
	}

	/**
	 * Returns [
	 *   'JobType' => [
	 *      'YYYY-MM-DD' => INT,
	 *      ...
	 *   ]
	 * ]
	 *
	 * @param string|null $jobTask
	 *
	 * @return array<string, array<string, mixed>>
	 */
	public function getFullStats(?string $jobTask = null): array {
		$driverName = $this->getDriverName();
		$fields = function (SelectQuery $query) use ($driverName) {
			$runtime = $query->expr('UNIX_TIMESTAMP(completed) - UNIX_TIMESTAMP(fetched)');
			switch ($driverName) {
				case static::DRIVER_SQLSERVER:
					$runtime = $query->expr("DATEDIFF(s, '1970-01-01 00:00:00', completed) - DATEDIFF(s, '1970-01-01 00:00:00', fetched)");

					break;
				case static::DRIVER_POSTGRES:
					$runtime = $query->expr('EXTRACT(EPOCH FROM completed) - EXTRACT(EPOCH FROM fetched)');

					break;
				case static::DRIVER_SQLITE:
					$runtime = $query->expr('julianday(completed) - julianday(fetched)');

					break;
			}

			return [
				'job_task',
				'created',
				'duration' => $runtime,
			];
		};

		$conditions = ['completed IS NOT' => null];
		if ($jobTask) {
			$conditions['job_task'] = $jobTask;
		}

		$jobs = $this->find()
			->select($fields)
			->where($conditions)
			->enableHydration(false)
			->orderByDesc('id')
			->limit(static::STATS_LIMIT)
			->all()
			->toArray();
		/** @var array<array{created: \DateTime, duration: int|float|null, job_task: string}> $jobs */

		$result = [];

		$days = [];

		foreach ($jobs as $job) {
			/** @var \DateTime $created */
			$created = $job['created'];
			$day = $created->format('Y-m-d');
			if (!isset($days[$day])) {
				$days[$day] = $day;
			}

			$runtime = $job['duration'];
			if ($driverName === static::DRIVER_SQLITE) {
				$runtime = (int)round($runtime * static::DAY);
			}

			/** @var string $name */
			$name = $job['job_task'];
			$result[$name][$day][] = $runtime;
		}

		foreach ($result as $jobTask => $jobs) {
			/**
			 * @var string $day
			 * @var array<int> $durations
			 */
			foreach ($jobs as $day => $durations) {
				$average = array_sum($durations) / count($durations);
				$result[$jobTask][$day] = (int)$average;
			}

			foreach ($days as $day) {
				if (isset($result[$jobTask][$day])) {
					continue;
				}

				$result[$jobTask][$day] = 0;
			}

			ksort($result[$jobTask]);
		}

		return $result;
	}

	/**
	 * Look for a new job that can be processed with the current abilities and
	 * from the specified group (or any if null).
	 *
	 * @param array<string, array<string, mixed>> $tasks Available QueueWorkerTasks.
	 * @param array<string> $groups Request a job from these groups (or exclude certain groups), or any otherwise.
	 * @param array<string> $types Request a job from these types (or exclude certain types), or any otherwise.
	 *
	 * @return \Queue\Model\Entity\QueuedJob|null
	 */
	public function requestJob(array $tasks, array $groups = [], array $types = []): ?QueuedJob {
		$now = $this->getDateTime();
		$nowStr = $now->toDateTimeString();
		$driverName = $this->getDriverName();

		$query = $this->find();
		$age = $query->expr()->add('IFNULL(TIMESTAMPDIFF(SECOND, "' . $nowStr . '", notbefore), 0)');
		switch ($driverName) {
			case static::DRIVER_SQLSERVER:
				$age = $query->expr()->add('ISNULL(DATEDIFF(SECOND, GETDATE(), notbefore), 0)');

				break;
			case static::DRIVER_POSTGRES:
				$age = $query->expr()
					->add('COALESCE(EXTRACT(EPOCH FROM notbefore) - (EXTRACT(EPOCH FROM now())), 0)');

				break;
			case static::DRIVER_SQLITE:
				$age = $query->expr()
					->add('IFNULL(CAST(strftime("%s", notbefore) as integer) - CAST(strftime("%s", "' . $nowStr . '") as integer), 0)');

				break;
		}
		$options = [
			'conditions' => [
				'completed IS' => null,
				'OR' => [],
			],
			'fields' => [
				'age' => $age,
			],
		];

		$costConstraints = [];
		foreach ($tasks as $name => $task) {
			if (!$task['costs']) {
				continue;
			}

			$costConstraints[$name] = $task['costs'];
		}

		$uniqueConstraints = [];
		foreach ($tasks as $name => $task) {
			if (!$task['unique']) {
				continue;
			}

			$uniqueConstraints[$name] = $name;
		}

		/** @var array<\Queue\Model\Entity\QueuedJob> $runningJobs */
		$runningJobs = [];
		if ($costConstraints || $uniqueConstraints) {
			$constraintJobs = array_keys($costConstraints + $uniqueConstraints);
			/** @var array<\Queue\Model\Entity\QueuedJob> $runningJobs */
			$runningJobs = $this->find('queued')
				->contain(['WorkerProcesses'])
				->where(['QueuedJobs.job_task IN' => $constraintJobs, 'QueuedJobs.workerkey IS NOT' => null, 'QueuedJobs.workerkey !=' => $this->_key, 'WorkerProcesses.modified >' => (new DateTime())->subSeconds(Config::defaultworkertimeout())])
				->all()
				->toArray();
		}

		$costs = 0;
		$server = $this->WorkerProcesses->buildServerString();
		foreach ($runningJobs as $runningJob) {
			if (isset($uniqueConstraints[$runningJob->job_task])) {
				$types[] = '-' . $runningJob->job_task;

				continue;
			}

			if ($runningJob->worker_process->server === $server && isset($costConstraints[$runningJob->job_task])) {
				$costs += $costConstraints[$runningJob->job_task];
			}
		}

		if ($costs) {
			$left = 100 - $costs;
			foreach ($tasks as $name => $task) {
				if (!$task['costs'] || $task['costs'] < $left) {
					continue;
				}

				$types[] = '-' . $name;
			}
		}

		if ($groups) {
			$options['conditions'] = $this->addFilter($options['conditions'], 'job_group', $groups);
		}
		if ($types) {
			$options['conditions'] = $this->addFilter($options['conditions'], 'job_task', $types);
		}

		// Generate the task specific conditions.
		foreach ($tasks as $name => $task) {
			$timeoutAt = clone $now;
			$tmp = [
				'job_task' => $name,
				'AND' => [
					[
						'OR' => [
							'notbefore <' => $nowStr,
							'notbefore IS' => null,
						],
					],
					[
						'OR' => [
							'fetched <' => $timeoutAt->subSeconds($task['timeout']),
							'fetched IS' => null,
						],
					],
				],
				'attempts <' => $task['retries'] + 1,
			];
			if (array_key_exists('rate', $task) && $tmp['job_task'] && array_key_exists($tmp['job_task'], $this->rateHistory)) {
				switch ($driverName) {
					case static::DRIVER_POSTGRES:
						$tmp['EXTRACT(EPOCH FROM NOW()) >='] = $this->rateHistory[$tmp['job_task']] + $task['rate'];

						break;
					case static::DRIVER_MYSQL:
						$tmp['UNIX_TIMESTAMP() >='] = $this->rateHistory[$tmp['job_task']] + $task['rate'];

						break;
					case static::DRIVER_SQLSERVER:
						$tmp["(DATEDIFF(s, '1970-01-01 00:00:00', GETDATE())) >="] = $this->rateHistory[$tmp['job_task']] + $task['rate'];

						break;
					case static::DRIVER_SQLITE:
						$tmp['strftime("%s", "now") >='] = $this->rateHistory[$tmp['job_task']] + $task['rate'];

						break;
				}
			}
			$options['conditions']['OR'][] = $tmp;
		}

		/** @var \Queue\Model\Entity\QueuedJob|null $job */
		$job = $this->getConnection()->transactional(function () use ($query, $options, $now, $driverName) {
			$query->find('all', ...$options)->enableAutoFields(true)
				->orderBy(['priority' => 'ASC', 'age' => 'ASC', 'id' => 'ASC']);

			switch ($driverName) {
				case static::DRIVER_MYSQL:
				case static::DRIVER_POSTGRES:
					$query->epilog('FOR UPDATE');

					break;
				case static::DRIVER_SQLSERVER:
					// Row-level locking (ROWLOCK, UPDLOCK hints) not supported by CakePHP ORM.
					// SQLServer uses default isolation level (READ COMMITTED) with automatic
					// row locking on UPDATE. This may allow race conditions in high-concurrency
					// scenarios. Consider using application-level locking or advisory locks if needed.

					break;
				case static::DRIVER_SQLITE:
					// not supported

					break;
			}

			/** @var \Queue\Model\Entity\QueuedJob|null $job */
			$job = $query->first();

			if (!$job) {
				return null;
			}

			$key = $this->key();
			$job = $this->patchEntity($job, [
				'workerkey' => $key,
				'fetched' => $now,
				'progress' => null,
				'failure_message' => null,
				'output' => null,
				'attempts' => $job->attempts + 1,
			]);

			return $this->saveOrFail($job);
		});

		if (!$job) {
			return null;
		}

		$this->rateHistory[$job->job_task] = $now->toUnixString();

		return $job;
	}

	/**
	 * @param int $id ID of job
	 * @param float $progress Value from 0 to 1
	 * @param string|null $status
	 *
	 * @return bool Success
	 */
	public function updateProgress(int $id, float $progress, ?string $status = null): bool {
		if (!$id) {
			return false;
		}

		$values = [
			'progress' => round($progress, 2),
			'memory' => Memory::usage(),
		];
		if ($status !== null) {
			$values['status'] = $status;
		}

		return (bool)$this->updateAll($values, ['id' => $id]);
	}

	/**
	 * Mark a job as Completed, removing it from the queue.
	 *
	 * @param \Queue\Model\Entity\QueuedJob $job Job
	 * @param string|null $output Optional captured output.
	 *
	 * @return bool Success
	 */
	public function markJobDone(QueuedJob $job, ?string $output = null): bool {
		$fields = [
			'progress' => 1,
			'completed' => $this->getDateTime(),
			'memory' => Memory::usage(),
		];
		if ($output !== null) {
			$fields['output'] = $output;
		}
		$job = $this->patchEntity($job, $fields);

		return (bool)$this->save($job);
	}

	/**
	 * Mark a job as Failed, without incrementing the "attempts" count due to to it being incremented when fetched.
	 *
	 * @param \Queue\Model\Entity\QueuedJob $job Job
	 * @param string|null $failureMessage Optional message to append to the failure_message field.
	 * @param string|null $output Optional captured output.
	 *
	 * @return bool Success
	 */
	public function markJobFailed(QueuedJob $job, ?string $failureMessage = null, ?string $output = null): bool {
		$fields = [
			'failure_message' => $failureMessage,
			'memory' => Memory::usage(),
		];
		if ($output !== null) {
			$fields['output'] = $output;
		}
		$job = $this->patchEntity($job, $fields);

		return (bool)$this->save($job);
	}

	/**
	 * Mark a job as terminally failed (aborted) after it has exhausted all
	 * retries. Records the terminal `status` so the job is no longer reported
	 * as queued/in-progress: `isQueued()` excludes it, the admin status filter
	 * skips it, and the status badge renders it as failed rather than running.
	 *
	 * Call this only once the worker has determined no further attempts will
	 * be made (see Processor, getFailedStatus() === self::STATUS_ABORTED).
	 *
	 * @param \Queue\Model\Entity\QueuedJob $job Job
	 *
	 * @return bool Success
	 */
	public function markJobAborted(QueuedJob $job): bool {
		return (bool)$this->updateAll(
			['status' => static::STATUS_ABORTED],
			['id' => $job->id],
		);
	}

	/**
	 * Removes all failed jobs.
	 *
	 * @return int Count of deleted rows
	 */
	public function flushFailedJobs(): int {
		$timeout = Config::defaultworkertimeout();
		$thresholdTime = (new DateTime())->subSeconds($timeout);

		$conditions = [
			'completed IS' => null,
			'attempts >' => 0,
			'fetched <' => $thresholdTime,
		];

		return $this->deleteAll($conditions);
	}

	/**
	 * Resets all failed and not yet completed jobs.
	 *
	 * @param int|null $id
	 * @param bool $full Also currently running jobs.
	 *
	 * @return int Success
	 */
	public function reset(?int $id = null, bool $full = false): int {
		$fields = [
			'completed' => null,
			'fetched' => null,
			'progress' => null,
			'attempts' => 0,
			'workerkey' => null,
			'failure_message' => null,
			'output' => null,
			'memory' => null,
			// Clear the terminal aborted status so a reset job counts as pending
			// again (getPendingCount()/isQueued() exclude status = aborted).
			'status' => null,
		];
		$conditions = [
			'completed IS' => null,
			'OR' => [
				'notbefore <=' => new DateTime(),
				'notbefore IS' => null,
			],
		];
		if ($id) {
			$conditions['id'] = $id;
		}
		if (!$full) {
			$conditions['attempts >'] = 0;
		}

		return $this->updateAll($fields, $conditions);
	}

	/**
	 * @param string $task
	 * @param string|null $reference
	 *
	 * @return int
	 */
	public function rerunByTask(string $task, ?string $reference = null): int {
		$fields = [
			'completed' => null,
			'fetched' => null,
			'progress' => null,
			'attempts' => 0,
			'workerkey' => null,
			'failure_message' => null,
			'output' => null,
			'memory' => null,
		];
		$conditions = [
			'completed IS NOT' => null,
			'job_task' => $task,
		];
		if ($reference) {
			$conditions['reference'] = $reference;
		}

		return $this->updateAll($fields, $conditions);
	}

	/**
	 * @param int $id
	 *
	 * @return int
	 */
	public function rerun(int $id): int {
		$fields = [
			'completed' => null,
			'fetched' => null,
			'progress' => null,
			'attempts' => 0,
			'workerkey' => null,
			'failure_message' => null,
			'output' => null,
			'memory' => null,
		];
		$conditions = [
			'completed IS NOT' => null,
			'id' => $id,
		];

		return $this->updateAll($fields, $conditions);
	}

	/**
	 * @param \Queue\Model\Entity\QueuedJob $queuedJob
	 *
	 * @return \Queue\Model\Entity\QueuedJob|null
	 */
	public function clone(QueuedJob $queuedJob): ?QueuedJob {
		$defaults = [
			'completed' => null,
			'fetched' => null,
			'progress' => null,
			'attempts' => 0,
			'workerkey' => null,
			'failure_message' => null,
			'output' => null,
			'memory' => null,
		];
		$data = $defaults + $queuedJob->toArray();
		$data['created'] = new DateTime();
		$queuedJob = $this->newEntity($data);

		return $this->save($queuedJob) ?: null;
	}

	/**
	 * Returns heatmap data aggregated by day of week and hour.
	 *
	 * @param string $field Field to aggregate on ('created' or 'completed')
	 * @param int $days Number of days to look back
	 * @param string|null $jobTask Filter by specific job task
	 *
	 * @return array{grid: array<int, array<int, int>>, summary: array<string, mixed>}
	 */
	public function getHeatmapData(string $field = 'created', int $days = 30, ?string $jobTask = null): array {
		// Whitelist allowed fields to prevent SQL injection
		$allowedFields = ['created', 'completed'];
		if (!in_array($field, $allowedFields, true)) {
			$field = 'created';
		}

		$driverName = $this->getDriverName();
		$since = (new DateTime())->subDays($days);

		$query = $this->find();

		// Build day of week and hour expressions based on driver
		switch ($driverName) {
			case static::DRIVER_POSTGRES:
				$dayOfWeek = $query->expr("EXTRACT(DOW FROM {$field})");
				$hourOfDay = $query->expr("EXTRACT(HOUR FROM {$field})");

				break;
			case static::DRIVER_SQLSERVER:
				// Use DATEDIFF against known Sunday (1900-01-07) for consistent day-of-week regardless of DATEFIRST setting
				$dayOfWeek = $query->expr("DATEDIFF(DAY, '1900-01-07', CAST({$field} AS DATE)) % 7");
				$hourOfDay = $query->expr("DATEPART(HOUR, {$field})");

				break;
			case static::DRIVER_SQLITE:
				$dayOfWeek = $query->expr("CAST(strftime('%w', {$field}) AS INTEGER)");
				$hourOfDay = $query->expr("CAST(strftime('%H', {$field}) AS INTEGER)");

				break;
			default: // MySQL
				$dayOfWeek = $query->expr("DAYOFWEEK({$field})");
				$hourOfDay = $query->expr("HOUR({$field})");

				break;
		}

		$conditions = [
			$field . ' IS NOT' => null,
			$field . ' >=' => $since,
		];
		if ($jobTask) {
			$conditions['job_task'] = $jobTask;
		}

		/** @var array<array{day_of_week: int, hour_of_day: int, job_count: int}> $results */
		$results = $query
			->select([
				'day_of_week' => $dayOfWeek,
				'hour_of_day' => $hourOfDay,
				'job_count' => $query->func()->count('*'),
			])
			->where($conditions)
			->groupBy(['day_of_week', 'hour_of_day'])
			->disableHydration()
			->toArray();

		// Initialize 7x24 grid (days x hours) with zeros
		$grid = [];
		for ($day = 0; $day < 7; $day++) {
			$grid[$day] = array_fill(0, 24, 0);
		}

		// Populate grid with results
		$totalJobs = 0;
		$peakCount = 0;
		$peakDay = 0;
		$peakHour = 0;

		foreach ($results as $row) {
			$day = (int)$row['day_of_week'];
			$hour = (int)$row['hour_of_day'];
			$count = (int)$row['job_count'];

			// MySQL DAYOFWEEK returns 1=Sunday, 2=Monday, etc. Adjust to 0=Sunday
			if ($driverName === static::DRIVER_MYSQL) {
				$day -= 1;
			}

			// Ensure day is in valid range (0-6)
			$day = max(0, min(6, $day));

			$grid[$day][$hour] = $count;
			$totalJobs += $count;

			if ($count > $peakCount) {
				$peakCount = $count;
				$peakDay = $day;
				$peakHour = $hour;
			}
		}

		// Calculate summary statistics
		$dayNames = ['Sunday', 'Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday'];
		$busiestDay = $dayNames[0];
		$dayCounts = [];
		foreach ($grid as $day => $hours) {
			$dayCounts[$day] = array_sum($hours);
		}
		$maxDayCount = 0;
		foreach ($dayCounts as $day => $count) {
			if ($count > $maxDayCount) {
				$maxDayCount = $count;
				$busiestDay = $dayNames[$day];
			}
		}

		// Find quietest hour
		$quietestCount = PHP_INT_MAX;
		$quietestDay = 0;
		$quietestHour = 0;
		foreach ($grid as $day => $hours) {
			foreach ($hours as $hour => $count) {
				if ($count < $quietestCount) {
					$quietestCount = $count;
					$quietestDay = $day;
					$quietestHour = $hour;
				}
			}
		}

		return [
			'grid' => $grid,
			'summary' => [
				'total' => $totalJobs,
				'avgPerHour' => $totalJobs > 0 ? round($totalJobs / ($days * 24), 1) : 0,
				'peakHour' => sprintf('%s %02d:00', $dayNames[$peakDay], $peakHour),
				'peakCount' => $peakCount,
				'quietestHour' => sprintf('%s %02d:00', $dayNames[$quietestDay], $quietestHour),
				'quietestCount' => $quietestCount === PHP_INT_MAX ? 0 : $quietestCount,
				'busiestDay' => $busiestDay,
				'days' => $days,
			],
		];
	}

	/**
	 * Return some statistics about unfinished jobs still in the Database.
	 *
	 * @return \Cake\ORM\Query\SelectQuery
	 */
	public function getPendingStats(): SelectQuery {
		$findCond = [
			'fields' => [
				'id',
				'job_task',
				'created',
				'status',
				'priority',
				'fetched',
				'progress',
				'reference',
				'notbefore',
				'attempts',
				'failure_message',
				'memory',
			],
			'conditions' => $this->pendingConditions(),
		];

		return $this->find('all', ...$findCond);
	}

	/**
	 * Shared WHERE conditions for "pending" jobs: not completed, due to run, and
	 * not terminally aborted. An aborted job keeps `completed IS NULL` but will
	 * never run again (retries exhausted), so it must not be reported as pending.
	 *
	 * @return array<int|string, mixed>
	 */
	protected function pendingConditions(): array {
		return [
			'completed IS' => null,
			[
				'OR' => [
					'notbefore <=' => new DateTime(),
					'notbefore IS' => null,
				],
			],
			[
				'OR' => [
					'status IS' => null,
					'status !=' => static::STATUS_ABORTED,
				],
			],
		];
	}

	/**
	 * Count of pending jobs (matches getPendingStats() conditions).
	 *
	 * Companion to getPendingStats() for callers that LIMIT the materialised
	 * details list and need the unbounded total to render "showing N of M".
	 *
	 * @return int
	 */
	public function getPendingCount(): int {
		return $this->find()
			->where($this->pendingConditions())
			->count();
	}

	/**
	 * Count of pending jobs that have never been fetched or attempted
	 * (the "new" tile on the admin dashboard).
	 *
	 * @return int
	 */
	public function getNewCount(): int {
		return $this->find()
			->where([
				'completed IS' => null,
				'fetched IS' => null,
				'attempts' => 0,
				'OR' => [
					'notbefore <=' => new DateTime(),
					'notbefore IS' => null,
				],
			])
			->count();
	}

	/**
	 * @return \Cake\ORM\Query\SelectQuery
	 */
	public function getScheduledStats(): SelectQuery {
		$findCond = [
			'fields' => [
				'id',
				'job_task',
				'created',
				'status',
				'priority',
				'fetched',
				'progress',
				'reference',
				'notbefore',
				'attempts',
				'failure_message',
				'memory',
			],
			'conditions' => [
				'completed IS' => null,
				'notbefore >' => new DateTime(),
			],
		];

		return $this->find('all', ...$findCond);
	}

	/**
	 * Count of scheduled jobs (matches getScheduledStats() conditions).
	 *
	 * Companion to getScheduledStats() for callers that LIMIT the
	 * materialised details list and need the unbounded total.
	 *
	 * @return int
	 */
	public function getScheduledCount(): int {
		return $this->find()
			->where([
				'completed IS' => null,
				'notbefore >' => new DateTime(),
			])
			->count();
	}

	/**
	 * Cleanup/Delete Completed Jobs.
	 *
	 * @return int
	 */
	public function cleanOldJobs(): int {
		$cleanupTimeout = Config::cleanuptimeout();
		if (!$cleanupTimeout) {
			return 0;
		}

		$threshold = (new DateTime())->subSeconds($cleanupTimeout);

		return $this->deleteAll([
			'completed <' => $threshold,
		]);
	}

	/**
	 * @param \Queue\Model\Entity\QueuedJob $queuedTask
	 * @param array<string, array<string, mixed>> $taskConfiguration
	 *
	 * @return string
	 */
	public function getFailedStatus(QueuedJob $queuedTask, array $taskConfiguration): string {
		$failureMessageRequeued = 'requeued';

		$queuedTaskName = $queuedTask->job_task;
		if (empty($taskConfiguration[$queuedTaskName])) {
			// Try with plugin prefix for backward compatibility (e.g. "Example" => "Queue.Example")
			$queuedTaskName = 'Queue.' . $queuedTask->job_task;
			if (empty($taskConfiguration[$queuedTaskName])) {
				return $failureMessageRequeued;
			}
		}

		$retries = $taskConfiguration[$queuedTaskName]['retries'];
		if ($queuedTask->attempts <= $retries) {
			return $failureMessageRequeued;
		}

		return static::STATUS_ABORTED;
	}

	/**
	 * Custom find method, as in `find('queued', ...)`.
	 *
	 * @param \Cake\ORM\Query\SelectQuery $query The query to find with
	 * @param array<string, mixed> $options The options to find with
	 *
	 * @return \Cake\ORM\Query\SelectQuery The query builder
	 */
	public function findQueued(SelectQuery $query, array $options = []): SelectQuery {
		return $query->where(['completed IS' => null]);
	}

	/**
	 * Generates a unique Identifier for the current worker thread.
	 *
	 * Useful to identify the currently running processes for this thread.
	 *
	 * @return string Identifier
	 */
	public function key(): string {
		if ($this->_key !== null) {
			return $this->_key;
		}
		$this->_key = bin2hex(random_bytes(20));

		return $this->_key;
	}

	/**
	 * Resets worker Identifier
	 *
	 * @return void
	 */
	public function clearKey(): void {
		$this->_key = null;
	}

	/**
	 * truncate()
	 *
	 * @return void
	 */
	public function truncate(): void {
		/** @var \Cake\Database\Schema\TableSchema $schema */
		$schema = $this->getSchema();
		$sql = $schema->truncateSql($this->getConnection());
		foreach ($sql as $snippet) {
			$this->getConnection()->execute($snippet);
		}
	}

	/**
	 * get the name of the driver
	 *
	 * @return string
	 */
	protected function getDriverName(): string {
		$className = explode('\\', (string)$this->getConnection()->config()['driver']);

		return end($className) ?: '';
	}

	/**
	 * @param array<mixed> $conditions
	 * @param string $key
	 * @param array<string> $values
	 *
	 * @return array<mixed>
	 */
	protected function addFilter(array $conditions, string $key, array $values): array {
		$include = [];
		$exclude = [];
		foreach ($values as $value) {
			if (str_starts_with($value, '-')) {
				$exclude[] = substr($value, 1);
			} else {
				$include[] = $value;
			}
		}

		if ($include) {
			$conditions[$key . ' IN'] = array_unique($include);
		}
		if ($exclude) {
			$conditions[$key . ' NOT IN'] = array_unique($exclude);
		}

		return $conditions;
	}

	/**
	 * Returns a DateTime object from different input.
	 *
	 * Without argument this will be "now".
	 *
	 * @param \Cake\I18n\DateTime|string|int|null $notBefore
	 *
	 * @return \Cake\I18n\DateTime
	 */
	protected function getDateTime(DateTime|string|int|null $notBefore = null): DateTime {
		if (is_object($notBefore)) {
			return $notBefore;
		}

		return new DateTime($notBefore);
	}

}