24850441053

Committed 23 Apr 2026 05:54PM UTC coverage: 52.404% (-1.5%) from 53.938%

Build # 24850441053

Build Type

push

github

Committed by

daycry

Commit Message

Fixes

Coverage Stats

104 of 219 new or added lines in 42 files covered. (47.49%)

14 existing lines in 9 files now uncovered.

1210 of 2309 relevant lines covered (52.4%)

4.37 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.45

/src/Queues/DatabaseQueue.php

<?php

declare(strict_types=1);

/**
 * This file is part of Daycry Queues.
 *
 * (c) Daycry <daycry9@proton.me>
 *
 * For the full copyright and license information, please view
 * the LICENSE file that was distributed with this source code.
 */

namespace Daycry\Jobs\Queues;

use Throwable;
use DateTime;
use DateTimeZone;
use Daycry\Jobs\Entities\Queue as QueueEntity;
use Daycry\Jobs\Interfaces\QueueInterface;
use Daycry\Jobs\Interfaces\WorkerInterface;
use Daycry\Jobs\Job as QueuesJob;
use Daycry\Jobs\Models\QueueModel;

/**
 * Database-backed queue (persistent) implementing scheduling & status tracking.
 * watch(): devuelve QueueEntity o null.
 */
class DatabaseQueue extends BaseQueue implements QueueInterface, WorkerInterface
{
    private mixed $job    = null;

    public function enqueue(object $data): string
    {
        helper('text');

        $identifier = $this->generateId(bytes: 4);

        $queueModel = new QueueModel();
        $job        = new QueueEntity();

        $delay = $this->calculateDelay($data);

        if ($delay->isImmediate()) {
            $data->schedule = new DateTime('now', new DateTimeZone(config('App')->appTimezone));
        }

        $data->identifier = $identifier;
        $job->queue       = $data->queue;
        $job->payload     = \json_encode($data);
        $job->priority    = $data->priority;
        $job->schedule    = $data->schedule->format('Y-m-d H:i:s');
        $job->identifier  = $identifier;
        $job->status      = 'pending';
        $job->max_retries = $data->maxRetries ?? null;
        $job->attempts    = $data->attempts ?? 0;

        $queueModel->insert($job);

        return $identifier;
    }

    public function watch(string $queue): ?JobEnvelope
    {
        $queueModel = new QueueModel();

        $this->job = $queueModel->reserveJob($queue);

        if ($this->job instanceof QueueEntity) {
            $decoded = \json_decode($this->job->payload ?? '{}');

            return JobEnvelope::fromBackend(
                backend: 'database',
                id: (string) $this->job->identifier,
                queue: (string) $this->job->queue,
                payload: $decoded,
                extraMeta: [
                    'entity_id' => $this->job->id,
                    'status'    => $this->job->status,
                    'schedule'  => $this->job->schedule ?? null,
                ],
                raw: $this->job,
            );
        }

        return null;
    }

    public function removeJob(QueuesJob $job, bool $recreate = false): bool
    {
        $queueModel = new QueueModel();

        if ($recreate) {
            // Update status and re-enqueue. If re-enqueue fails, restore previous status.
            $previousStatus = null;
            if ($this->job !== null) {
                $previousStatus    = $this->job->status;
                $this->job->status = 'failed';
                $queueModel->update($this->job->id, $this->job);
            }
            try {
                $this->enqueue($job->toObject());
            } catch (Throwable $e) {
                // Rollback status change if re-enqueue fails
                if ($this->job !== null && $previousStatus !== null) {
                    $this->job->status = $previousStatus;
                    $queueModel->update($this->job->id, $this->job);
                }
                log_message('error', 'DatabaseQueue::removeJob re-enqueue failed: ' . $e->getMessage());

                throw $e;
            }
        } elseif ($this->job !== null) {
            // Completion: mark as completed
            $this->job->status = 'completed';
            $queueModel->update($this->job->id, $this->job);
        }

        $this->job = null;

        return true;
    }

    public function setPriority(int $priority)
    {
        return $this;
    }
}

1	<?php
2
3	declare(strict_types=1);
4
5	/**
6	* This file is part of Daycry Queues.
7	*
8	* (c) Daycry <daycry9@proton.me>
9	*
10	* For the full copyright and license information, please view
11	* the LICENSE file that was distributed with this source code.
12	*/
13
14	namespace Daycry\Jobs\Queues;
15
16	use Throwable;
17	use DateTime;
18	use DateTimeZone;
19	use Daycry\Jobs\Entities\Queue as QueueEntity;
20	use Daycry\Jobs\Interfaces\QueueInterface;
21	use Daycry\Jobs\Interfaces\WorkerInterface;
22	use Daycry\Jobs\Job as QueuesJob;
23	use Daycry\Jobs\Models\QueueModel;
24
25	/**
26	* Database-backed queue (persistent) implementing scheduling & status tracking.
27	* watch(): devuelve QueueEntity o null.
28	*/
29	class DatabaseQueue extends BaseQueue implements QueueInterface, WorkerInterface
30	{
31	private mixed $job = null;
32
33	public function enqueue(object $data): string
34	{
35	helper('text');	6✔
36
37	$identifier = $this->generateId(bytes: 4);	6✔
38
39	$queueModel = new QueueModel();	6✔
40	$job = new QueueEntity();	6✔
41
42	$delay = $this->calculateDelay($data);	6✔
43
44	if ($delay->isImmediate()) {	6✔
45	$data->schedule = new DateTime('now', new DateTimeZone(config('App')->appTimezone));	6✔
46	}
47
48	$data->identifier = $identifier;	6✔
49	$job->queue = $data->queue;	6✔
50	$job->payload = \json_encode($data);	6✔
51	$job->priority = $data->priority;	6✔
52	$job->schedule = $data->schedule->format('Y-m-d H:i:s');	6✔
53	$job->identifier = $identifier;	6✔
54	$job->status = 'pending';	6✔
55	$job->max_retries = $data->maxRetries ?? null;	6✔
56	$job->attempts = $data->attempts ?? 0;	6✔
57
58	$queueModel->insert($job);	6✔
59
60	return $identifier;	6✔
61	}
62
63	public function watch(string $queue): ?JobEnvelope
64	{
65	$queueModel = new QueueModel();	3✔
66
67	$this->job = $queueModel->reserveJob($queue);	3✔
68
69	if ($this->job instanceof QueueEntity) {	3✔
70	$decoded = \json_decode($this->job->payload ?? '{}');	3✔
71
72	return JobEnvelope::fromBackend(	3✔
73	backend: 'database',	3✔
74	id: (string) $this->job->identifier,	3✔
75	queue: (string) $this->job->queue,	3✔
76	payload: $decoded,	3✔
77	extraMeta: [	3✔
78	'entity_id' => $this->job->id,	3✔
79	'status' => $this->job->status,	3✔
80	'schedule' => $this->job->schedule ?? null,	3✔
81	],	3✔
82	raw: $this->job,	3✔
83	);	3✔
84	}
85
86	return null;	×
87	}
88
89	public function removeJob(QueuesJob $job, bool $recreate = false): bool
90	{
91	$queueModel = new QueueModel();	2✔
92
93	if ($recreate) {	2✔
94	// Update status and re-enqueue. If re-enqueue fails, restore previous status.
95	$previousStatus = null;	1✔
96	if ($this->job !== null) {	1✔
97	$previousStatus = $this->job->status;	1✔
98	$this->job->status = 'failed';	1✔
99	$queueModel->update($this->job->id, $this->job);	1✔
100	}
101	try {
102	$this->enqueue($job->toObject());	1✔
NEW 103	} catch (Throwable $e) {	×
104	// Rollback status change if re-enqueue fails
NEW 105	if ($this->job !== null && $previousStatus !== null) {	×
NEW 106	$this->job->status = $previousStatus;	×
NEW 107	$queueModel->update($this->job->id, $this->job);	×
108	}
NEW 109	log_message('error', 'DatabaseQueue::removeJob re-enqueue failed: ' . $e->getMessage());	×
110
NEW 111	throw $e;	×
112	}
113	} elseif ($this->job !== null) {	1✔
114	// Completion: mark as completed
115	$this->job->status = 'completed';	1✔
116	$queueModel->update($this->job->id, $this->job);	1✔
117	}
118
119	$this->job = null;	2✔
120
121	return true;	2✔
122	}
123
124	public function setPriority(int $priority)
125	{
126	return $this;	×
127	}
128	}

daycry / jobs / 24850441053

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous