CronJob.php — Download Manager 3.3.63

download-manager / src / __ / CronJob.php

download-manager / src / __ Last commit date

HTML 1 year ago Jobs 2 days ago views 2 days ago ActivityReport.php 2 days ago Apply.php 6 months ago Cron.php 1 year ago CronJob.php 2 days ago CronJobs.php 2 days ago Crypt.php 2 months ago DownloadStats.php 5 months ago Email.php 2 days ago EmailCron.php 1 year ago FileSystem.php 1 year ago Installer.php 2 days ago Messages.php 1 year ago Query.php 5 months ago Session.php 1 week ago Settings.php 5 years ago SimpleMath.php 5 years ago TempStorage.php 1 week ago Template.php 5 months ago UI.php 7 months ago Updater.php 4 years ago UserAgent.php 2 years ago __.php 2 months ago __MailUI.php 3 years ago

CronJob.php

765 lines

1	<?php
2	/**
3	* WPDM Cron Job Manager
4	*
5	* Manages scheduled jobs with support for queues, priorities, retries, and locking.
6	*
7	* @package WPDM\__
8	* @since 7.0.1
9	*/
10
11	namespace WPDM\__;
12
13	use WPDM\__\Jobs\JobBuilder;
14	use WPDM\__\Jobs\JobHandler;
15
16	class CronJob
17	{
18	/**
19	* Singleton instance
20	*
21	* @var self\|null
22	*/
23	private static $instance;
24
25	/**
26	* Registered job handlers (whitelist for security)
27	*
28	* @var array
29	*/
30	private static $registeredHandlers = [];
31
32	/**
33	* Current worker ID for job locking
34	*
35	* @var string\|null
36	*/
37	private static $workerId;
38
39	/**
40	* Job statuses
41	*/
42	const STATUS_PENDING = 'pending';
43	const STATUS_RUNNING = 'running';
44	const STATUS_COMPLETED = 'completed';
45	const STATUS_FAILED = 'failed';
46	const STATUS_CANCELLED = 'cancelled';
47
48	/**
49	* Get singleton instance
50	*
51	* @return self
52	*/
53	public static function getInstance(): self
54	{
55	if (self::$instance === null) {
56	self::$instance = new self();
57	}
58	return self::$instance;
59	}
60
61	/**
62	* Constructor - register default handlers
63	*/
64	public function __construct()
65	{
66	self::$workerId = wp_generate_uuid4();
67	$this->registerDefaultHandlers();
68	}
69
70	/**
71	* Register default job handlers
72	*
73	* @return void
74	*/
75	private function registerDefaultHandlers(): void
76	{
77	// Register built-in handlers
78	$builtInHandlers = [
79	\WPDM\__\Jobs\CleanupJob::class,
80	\WPDM\__\Jobs\DeleteExpiredJob::class,
81	\WPDM\__\Jobs\SendEmailJob::class,
82	\WPDM\__\Jobs\ActivityReportJob::class,
83	];
84
85	foreach ($builtInHandlers as $handler) {
86	if (class_exists($handler)) {
87	self::registerHandler($handler);
88	}
89	}
90
91	// Allow add-ons to register their handlers
92	do_action('wpdm_register_job_handlers');
93	}
94
95	/**
96	* Register a job handler class (whitelist)
97	*
98	* @param string $handlerClass Fully qualified class name
99	* @return void
100	*/
101	public static function registerHandler(string $handlerClass): void
102	{
103	if (!in_array($handlerClass, self::$registeredHandlers, true)) {
104	self::$registeredHandlers[] = $handlerClass;
105	}
106	}
107
108	/**
109	* Check if a handler is registered (whitelisted)
110	*
111	* @param string $handlerClass
112	* @return bool
113	*/
114	public static function isHandlerRegistered(string $handlerClass): bool
115	{
116	return in_array($handlerClass, self::$registeredHandlers, true);
117	}
118
119	/**
120	* Get all registered handlers
121	*
122	* @return array
123	*/
124	public static function getRegisteredHandlers(): array
125	{
126	return self::$registeredHandlers;
127	}
128
129	/**
130	* Get the cron key for authentication
131	*
132	* @return string
133	*/
134	public function cronKey(): string
135	{
136	$key = get_option('__wpdm_cron_key');
137	if (!$key) {
138	$key = wp_generate_password(32, false);
139	update_option('__wpdm_cron_key', $key);
140	}
141	return $key;
142	}
143
144	/**
145	* Create a job builder for fluent API
146	*
147	* Usage:
148	* CronJob::dispatch(MyJob::class)->withData(['foo' => 'bar'])->delay(3600)->create();
149	*
150	* @param string $handler Job handler class name
151	* @return JobBuilder
152	*/
153	public static function dispatch(string $handler): JobBuilder
154	{
155	return new JobBuilder($handler);
156	}
157
158	/**
159	* Create a new job
160	*
161	* @param string $type Job handler class name
162	* @param array $data Job payload
163	* @param int $executeAt Unix timestamp when to execute
164	* @param int $repeatExecution Number of times to repeat (0 = infinite)
165	* @param int $interval Seconds between repeats
166	* @param string $queue Queue name
167	* @param int $priority Priority (1-10)
168	* @param int $maxAttempts Maximum retry attempts
169	* @param string\|null $uniqueCode Unique code to prevent duplicates
170	* @return int\|false Job ID on success, false on failure
171	*/
172	public static function create(
173	string $type,
174	array $data = [],
175	int $executeAt = 0,
176	int $repeatExecution = 1,
177	int $interval = 0,
178	string $queue = 'default',
179	int $priority = 5,
180	int $maxAttempts = 3,
181	?string $uniqueCode = null
182	) {
183	global $wpdb;
184
185	// Validate handler is registered
186	if (!self::isHandlerRegistered($type)) {
187	error_log("[WPDM CronJob] Attempted to create job with unregistered handler: {$type}");
188	return false;
189	}
190
191	// Generate unique code if not provided
192	if ($uniqueCode === null) {
193	$uniqueCode = md5($type . $executeAt . wp_json_encode($data) . microtime(true));
194	}
195
196	// Check for duplicate job with same unique code
197	$existing = $wpdb->get_var($wpdb->prepare(
198	"SELECT ID FROM {$wpdb->prefix}ahm_cron_jobs WHERE code = %s AND status IN ('pending', 'running')",
199	$uniqueCode
200	));
201
202	if ($existing) {
203	return (int) $existing; // Return existing job ID
204	}
205
206	$executeAt = $executeAt ?: time();
207
208	$result = $wpdb->insert(
209	"{$wpdb->prefix}ahm_cron_jobs",
210	[
211	'code' => $uniqueCode,
212	'type' => $type,
213	'queue' => $queue,
214	'priority' => max(1, min(10, $priority)),
215	'data' => wp_json_encode($data),
216	'status' => self::STATUS_PENDING,
217	'attempts' => 0,
218	'max_attempts' => max(1, $maxAttempts),
219	'execute_at' => $executeAt,
220	'repeat_execution' => max(0, $repeatExecution),
221	'execution_count' => 0,
222	'interval_seconds' => max(0, $interval),
223	'created_at' => time(),
224	'created_by' => get_current_user_id(),
225	],
226	['%s', '%s', '%s', '%d', '%s', '%s', '%d', '%d', '%d', '%d', '%d', '%d', '%d', '%d']
227	);
228
229	if ($result === false) {
230	error_log("[WPDM CronJob] Failed to create job: " . $wpdb->last_error);
231	return false;
232	}
233
234	return $wpdb->insert_id;
235	}
236
237	/**
238	* Delete a job by ID
239	*
240	* @param int $id Job ID
241	* @return bool
242	*/
243	public static function delete(int $id): bool
244	{
245	global $wpdb;
246	return $wpdb->delete(
247	"{$wpdb->prefix}ahm_cron_jobs",
248	['ID' => $id],
249	['%d']
250	) !== false;
251	}
252
253	/**
254	* Cancel a pending job
255	*
256	* @param int $id Job ID
257	* @return bool
258	*/
259	public static function cancel(int $id): bool
260	{
261	global $wpdb;
262	return $wpdb->update(
263	"{$wpdb->prefix}ahm_cron_jobs",
264	['status' => self::STATUS_CANCELLED],
265	['ID' => $id, 'status' => self::STATUS_PENDING],
266	['%s'],
267	['%d', '%s']
268	) !== false;
269	}
270
271	/**
272	* Retry a failed job
273	*
274	* @param int $id Job ID
275	* @return bool
276	*/
277	public static function retry(int $id): bool
278	{
279	global $wpdb;
280	return $wpdb->update(
281	"{$wpdb->prefix}ahm_cron_jobs",
282	[
283	'status' => self::STATUS_PENDING,
284	'attempts' => 0,
285	'execute_at' => time(),
286	'error_message' => null,
287	'locked_by' => null,
288	'locked_at' => null,
289	],
290	['ID' => $id],
291	['%s', '%d', '%d', '%s', '%s', '%s'],
292	['%d']
293	) !== false;
294	}
295
296	/**
297	* Execute all pending jobs (limited batch)
298	*
299	* @param int $limit Maximum jobs to process
300	* @param string\|null $queue Specific queue to process (null = all)
301	* @return array Execution results
302	*/
303	public function executeAll(int $limit = 10, ?string $queue = null): array
304	{
305	global $wpdb;
306
307	$results = [
308	'processed' => 0,
309	'succeeded' => 0,
310	'failed' => 0,
311	'jobs' => [],
312	];
313
314	$time = time();
315
316	// Build query
317	$sql = "SELECT * FROM {$wpdb->prefix}ahm_cron_jobs
318	WHERE status = %s
319	AND execute_at <= %d
320	AND locked_by IS NULL";
321	$params = [self::STATUS_PENDING, $time];
322
323	if ($queue !== null) {
324	$sql .= " AND queue = %s";
325	$params[] = $queue;
326	}
327
328	$sql .= " ORDER BY priority DESC, execute_at ASC LIMIT %d";
329	$params[] = $limit;
330
331	$jobs = $wpdb->get_results($wpdb->prepare($sql, ...$params));
332
333	foreach ($jobs as $job) {
334	$result = $this->execute($job);
335	$results['processed']++;
336
337	if ($result['success']) {
338	$results['succeeded']++;
339	} else {
340	$results['failed']++;
341	}
342
343	$results['jobs'][] = $result;
344	}
345
346	return $results;
347	}
348
349	/**
350	* Execute a single job
351	*
352	* @param object\|int $job Job object or ID
353	* @return array Result with 'success', 'message', 'job_id' keys
354	*/
355	public function execute($job): array
356	{
357	global $wpdb;
358
359	// Fetch job if ID provided
360	if (is_int($job) \|\| is_numeric($job)) {
361	$job = $wpdb->get_row($wpdb->prepare(
362	"SELECT * FROM {$wpdb->prefix}ahm_cron_jobs WHERE ID = %d",
363	(int) $job
364	));
365	}
366
367	if (!$job) {
368	return ['success' => false, 'message' => 'Job not found', 'job_id' => 0];
369	}
370
371	// Try to acquire lock
372	if (!$this->acquireLock($job)) {
373	return ['success' => false, 'message' => 'Could not acquire lock', 'job_id' => $job->ID];
374	}
375
376	// Validate handler is registered
377	if (!self::isHandlerRegistered($job->type)) {
378	$this->markFailed($job, "Unregistered handler: {$job->type}");
379	return ['success' => false, 'message' => 'Unregistered handler', 'job_id' => $job->ID];
380	}
381
382	// Validate handler class exists
383	if (!class_exists($job->type)) {
384	$this->markFailed($job, "Handler class not found: {$job->type}");
385	return ['success' => false, 'message' => 'Handler class not found', 'job_id' => $job->ID];
386	}
387
388	// Execute the job
389	try {
390	$data = json_decode($job->data);
391	$handler = new $job->type($data, $job);
392
393	if (!($handler instanceof JobHandler)) {
394	throw new \Exception("Handler must implement JobHandler interface");
395	}
396
397	$success = $handler->handle($data);
398
399	if ($success) {
400	$this->markCompleted($job);
401	return ['success' => true, 'message' => 'Job completed', 'job_id' => $job->ID];
402	} else {
403	throw new \Exception("Job handler returned false");
404	}
405	} catch (\Throwable $e) {
406	return $this->handleFailure($job, $e);
407	}
408	}
409
410	/**
411	* Acquire a lock on a job to prevent concurrent execution
412	*
413	* @param object $job
414	* @return bool
415	*/
416	private function acquireLock(object $job): bool
417	{
418	global $wpdb;
419
420	$affected = $wpdb->query($wpdb->prepare(
421	"UPDATE {$wpdb->prefix}ahm_cron_jobs
422	SET locked_by = %s, locked_at = %d, status = %s, started_at = %d
423	WHERE ID = %d AND locked_by IS NULL AND status = %s",
424	self::$workerId,
425	time(),
426	self::STATUS_RUNNING,
427	time(),
428	$job->ID,
429	self::STATUS_PENDING
430	));
431
432	return $affected > 0;
433	}
434
435	/**
436	* Release lock on a job
437	*
438	* @param object $job
439	* @return void
440	*/
441	private function releaseLock(object $job): void
442	{
443	global $wpdb;
444
445	$wpdb->update(
446	"{$wpdb->prefix}ahm_cron_jobs",
447	['locked_by' => null, 'locked_at' => null],
448	['ID' => $job->ID],
449	['%s', '%s'],
450	['%d']
451	);
452	}
453
454	/**
455	* Mark job as completed
456	*
457	* @param object $job
458	* @return void
459	*/
460	private function markCompleted(object $job): void
461	{
462	global $wpdb;
463
464	$executionCount = $job->execution_count + 1;
465	$repeatExecution = (int) $job->repeat_execution;
466
467	// Check if job should repeat
468	if ($repeatExecution === 0 \|\| $executionCount < $repeatExecution) {
469	// Schedule next execution
470	$nextExecuteAt = time() + (int) $job->interval_seconds;
471
472	$wpdb->update(
473	"{$wpdb->prefix}ahm_cron_jobs",
474	[
475	'status' => self::STATUS_PENDING,
476	'execution_count' => $executionCount,
477	'execute_at' => $nextExecuteAt,
478	'attempts' => 0,
479	'locked_by' => null,
480	'locked_at' => null,
481	'completed_at' => time(),
482	'error_message' => null,
483	],
484	['ID' => $job->ID],
485	['%s', '%d', '%d', '%d', '%s', '%s', '%d', '%s'],
486	['%d']
487	);
488	} else {
489	// Job fully completed
490	$wpdb->update(
491	"{$wpdb->prefix}ahm_cron_jobs",
492	[
493	'status' => self::STATUS_COMPLETED,
494	'execution_count' => $executionCount,
495	'locked_by' => null,
496	'locked_at' => null,
497	'completed_at' => time(),
498	],
499	['ID' => $job->ID],
500	['%s', '%d', '%s', '%s', '%d'],
501	['%d']
502	);
503	}
504	}
505
506	/**
507	* Handle job failure with retry logic
508	*
509	* @param object $job
510	* @param \Throwable $e
511	* @return array
512	*/
513	private function handleFailure(object $job, \Throwable $e): array
514	{
515	global $wpdb;
516
517	$attempts = (int) $job->attempts + 1;
518	$maxAttempts = (int) $job->max_attempts;
519
520	// Notify handler of failure
521	try {
522	if (class_exists($job->type)) {
523	$data = json_decode($job->data);
524	$handler = new $job->type($data, $job);
525	if ($handler instanceof JobHandler) {
526	$handler->failed($e, $data);
527	}
528	}
529	} catch (\Throwable $handlerError) {
530	error_log("[WPDM CronJob] Error in failure handler: " . $handlerError->getMessage());
531	}
532
533	$errorMessage = sprintf(
534	"%s in %s:%d\nStack trace:\n%s",
535	$e->getMessage(),
536	$e->getFile(),
537	$e->getLine(),
538	$e->getTraceAsString()
539	);
540
541	if ($attempts < $maxAttempts) {
542	// Schedule retry with exponential backoff
543	$retryDelay = min(3600, pow(2, $attempts) * 60); // Max 1 hour
544	$nextRetryAt = time() + $retryDelay;
545
546	$wpdb->update(
547	"{$wpdb->prefix}ahm_cron_jobs",
548	[
549	'status' => self::STATUS_PENDING,
550	'attempts' => $attempts,
551	'execute_at' => $nextRetryAt,
552	'next_retry_at' => $nextRetryAt,
553	'locked_by' => null,
554	'locked_at' => null,
555	'error_message' => $errorMessage,
556	],
557	['ID' => $job->ID],
558	['%s', '%d', '%d', '%d', '%s', '%s', '%s'],
559	['%d']
560	);
561
562	return [
563	'success' => false,
564	'message' => "Job failed, retry {$attempts}/{$maxAttempts} scheduled",
565	'job_id' => $job->ID,
566	];
567	} else {
568	// Max attempts reached, mark as failed
569	$this->markFailed($job, $errorMessage);
570
571	return [
572	'success' => false,
573	'message' => "Job failed after {$attempts} attempts",
574	'job_id' => $job->ID,
575	];
576	}
577	}
578
579	/**
580	* Mark job as permanently failed
581	*
582	* @param object $job
583	* @param string $errorMessage
584	* @return void
585	*/
586	private function markFailed(object $job, string $errorMessage): void
587	{
588	global $wpdb;
589
590	$wpdb->update(
591	"{$wpdb->prefix}ahm_cron_jobs",
592	[
593	'status' => self::STATUS_FAILED,
594	'locked_by' => null,
595	'locked_at' => null,
596	'error_message' => $errorMessage,
597	'completed_at' => time(),
598	],
599	['ID' => $job->ID],
600	['%s', '%s', '%s', '%s', '%d'],
601	['%d']
602	);
603	}
604
605	/**
606	* Get all jobs with optional filtering
607	*
608	* @param array $args Filter arguments: status, queue, limit, offset
609	* @return array
610	*/
611	public static function getAll(array $args = []): array
612	{
613	global $wpdb;
614
615	$defaults = [
616	'status' => null,
617	'queue' => null,
618	'limit' => 100,
619	'offset' => 0,
620	'orderby' => 'created_at',
621	'order' => 'DESC',
622	];
623
624	$args = wp_parse_args($args, $defaults);
625
626	$sql = "SELECT * FROM {$wpdb->prefix}ahm_cron_jobs WHERE 1=1";
627	$params = [];
628
629	if ($args['status']) {
630	$sql .= " AND status = %s";
631	$params[] = $args['status'];
632	}
633
634	if ($args['queue']) {
635	$sql .= " AND queue = %s";
636	$params[] = $args['queue'];
637	}
638
639	$allowedOrderBy = ['created_at', 'execute_at', 'priority', 'ID'];
640	$orderby = in_array($args['orderby'], $allowedOrderBy) ? $args['orderby'] : 'created_at';
641	$order = strtoupper($args['order']) === 'ASC' ? 'ASC' : 'DESC';
642
643	$sql .= " ORDER BY {$orderby} {$order}";
644	$sql .= " LIMIT %d OFFSET %d";
645	$params[] = (int) $args['limit'];
646	$params[] = (int) $args['offset'];
647
648	if (!empty($params)) {
649	$sql = $wpdb->prepare($sql, ...$params);
650	}
651
652	return $wpdb->get_results($sql);
653	}
654
655	/**
656	* Get job by ID
657	*
658	* @param int $id
659	* @return object\|null
660	*/
661	public static function get(int $id): ?object
662	{
663	global $wpdb;
664
665	return $wpdb->get_row($wpdb->prepare(
666	"SELECT * FROM {$wpdb->prefix}ahm_cron_jobs WHERE ID = %d",
667	$id
668	));
669	}
670
671	/**
672	* Get job counts by status
673	*
674	* @return array
675	*/
676	public static function getCounts(): array
677	{
678	global $wpdb;
679
680	$results = $wpdb->get_results(
681	"SELECT status, COUNT(*) as count FROM {$wpdb->prefix}ahm_cron_jobs GROUP BY status"
682	);
683
684	$counts = [
685	'total' => 0,
686	self::STATUS_PENDING => 0,
687	self::STATUS_RUNNING => 0,
688	self::STATUS_COMPLETED => 0,
689	self::STATUS_FAILED => 0,
690	self::STATUS_CANCELLED => 0,
691	];
692
693	foreach ($results as $row) {
694	$counts[$row->status] = (int) $row->count;
695	$counts['total'] += (int) $row->count;
696	}
697
698	return $counts;
699	}
700
701	/**
702	* Clean up old completed/cancelled/failed jobs
703	*
704	* @param int $olderThanDays Delete jobs older than this many days
705	* @return int Number of deleted jobs
706	*/
707	public static function cleanup(int $olderThanDays = 30): int
708	{
709	global $wpdb;
710
711	$cutoff = time() - ($olderThanDays * 86400);
712
713	return $wpdb->query($wpdb->prepare(
714	"DELETE FROM {$wpdb->prefix}ahm_cron_jobs
715	WHERE status IN (%s, %s, %s) AND completed_at < %d",
716	self::STATUS_COMPLETED,
717	self::STATUS_FAILED,
718	self::STATUS_CANCELLED,
719	$cutoff
720	));
721	}
722
723	/**
724	* Release stale locks (jobs that have been running too long)
725	*
726	* @param int $staleLockMinutes Minutes after which a lock is considered stale
727	* @return int Number of released locks
728	*/
729	public static function releaseStale(int $staleLockMinutes = 30): int
730	{
731	global $wpdb;
732
733	$cutoff = time() - ($staleLockMinutes * 60);
734
735	return $wpdb->query($wpdb->prepare(
736	"UPDATE {$wpdb->prefix}ahm_cron_jobs
737	SET status = %s, locked_by = NULL, locked_at = NULL
738	WHERE status = %s AND locked_at < %d",
739	self::STATUS_PENDING,
740	self::STATUS_RUNNING,
741	$cutoff
742	));
743	}
744
745	/**
746	* Pause all jobs in a queue
747	*
748	* @param string $queue
749	* @return int Number of affected jobs
750	*/
751	public static function pauseQueue(string $queue): int
752	{
753	global $wpdb;
754
755	return $wpdb->query($wpdb->prepare(
756	"UPDATE {$wpdb->prefix}ahm_cron_jobs
757	SET status = %s
758	WHERE queue = %s AND status = %s",
759	self::STATUS_CANCELLED,
760	$queue,
761	self::STATUS_PENDING
762	));
763	}
764	}
765