sync-basalam / JobManager.php

JobManager.php in ووسلام – همگام سازی ووکامرس و باسلام 1.10.12, at JobManager.php

316 lines 12.1 KB

Raw Download Zip

1	<?php
2
3	namespace SyncBasalam;
4
5	defined('ABSPATH') \|\| exit;
6
7	class JobManager
8	{
9	private $jobManagerTableName;
10
11	private const ALLOWED_COLUMNS = [
12	'id', 'job_type', 'status', 'payload',
13	'attempts', 'max_attempts', 'retry_after',
14	'started_at', 'created_at', 'failed_at', 'error_message',
15	];
16
17	function __construct()
18	{
19	global $wpdb;
20	$this->jobManagerTableName = $wpdb->prefix . 'sync_basalam_job_manager';
21	}
22
23	public function createJob($jobType, $status = 'pending', $payload = null, $maxAttempts = 3)
24	{
25	global $wpdb;
26
27	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Custom plugin table; no object cache for these operational queries.
28	$result = $wpdb->insert(
29	$this->jobManagerTableName,
30	array(
31	'job_type' => $jobType,
32	'status' => $status,
33	'payload' => $payload,
34	'attempts' => 0,
35	'max_attempts' => $maxAttempts,
36	'created_at' => time(),
37	)
38	);
39
40	if ($result !== false) {
41	do_action('sync_basalam_job_created', $jobType, $status, $payload);
42	}
43
44	return $result;
45	}
46
47	public function getNextEligibleJob(string $jobType): ?object
48	{
49	global $wpdb;
50
51	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, PluginCheck.Security.DirectDB.UnescapedDBParameter -- Custom plugin table; identifier from $wpdb->prefix, not user input.
52	return $wpdb->get_row($wpdb->prepare(
53	"SELECT * FROM {$this->jobManagerTableName}
54	WHERE job_type = %s
55	AND status = 'pending'
56	AND (retry_after IS NULL OR retry_after <= %d)
57	ORDER BY id ASC
58	LIMIT 1",
59	$jobType,
60	time()
61	));
62	}
63
64	public function hasAnyProcessingJob(): bool
65	{
66	return $this->getCountJobs(['status' => 'processing']) > 0;
67	}
68
69	public function hasPendingOrStaleProcessingJobs(int $staleProcessingTimeoutSeconds = 120): bool
70	{
71	global $wpdb;
72
73	$now = time();
74	$staleBefore = $now - $staleProcessingTimeoutSeconds;
75
76	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, PluginCheck.Security.DirectDB.UnescapedDBParameter -- Custom plugin table; identifier from $wpdb->prefix, not user input.
77	$result = $wpdb->get_var($wpdb->prepare(
78	"SELECT 1 FROM {$this->jobManagerTableName}
79	WHERE (
80	status = 'pending'
81	AND (retry_after IS NULL OR retry_after <= %d)
82	) OR (
83	status = 'processing'
84	AND started_at IS NOT NULL
85	AND started_at < %d
86	)
87	LIMIT 1",
88	$now,
89	$staleBefore
90	));
91
92	return (string) $result === '1';
93	}
94
95	public function getJob($where = array())
96	{
97	global $wpdb;
98
99	if (empty($where)) return null;
100
101	$conditions = [];
102	$values = [];
103
104	foreach ($where as $column => $value) {
105	if (!in_array($column, self::ALLOWED_COLUMNS, true)) {
106	throw new \InvalidArgumentException(esc_html("Invalid column: {$column}"));
107	}
108	$conditions[] = "{$column} = %s";
109	$values[] = $value;
110	}
111
112	$sql = "SELECT * FROM {$this->jobManagerTableName} WHERE " . implode(" AND ", $conditions) . " LIMIT 1";
113
114	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, PluginCheck.Security.DirectDB.UnescapedDBParameter -- Custom plugin table; identifiers from $wpdb->prefix and whitelisted column list, not user input; values are prepared.
115	return $wpdb->get_row($wpdb->prepare($sql, $values));
116	}
117
118	public function getCountJobs($where = array())
119	{
120	global $wpdb;
121
122	if (empty($where)) return 0;
123
124	$conditions = [];
125	$values = [];
126
127	foreach ($where as $column => $value) {
128	if (!in_array($column, self::ALLOWED_COLUMNS, true)) {
129	throw new \InvalidArgumentException(esc_html("Invalid column: {$column}"));
130	}
131	if (is_array($value)) {
132	$placeholders = array_fill(0, count($value), '%s');
133	$conditions[] = "{$column} IN (" . implode(',', $placeholders) . ")";
134	$values = array_merge($values, $value);
135	} else {
136	$conditions[] = "{$column} = %s";
137	$values[] = $value;
138	}
139	}
140
141	$sql = "SELECT COUNT(*) FROM {$this->jobManagerTableName} WHERE " . implode(" AND ", $conditions);
142
143	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, PluginCheck.Security.DirectDB.UnescapedDBParameter -- Custom plugin table; identifiers from $wpdb->prefix and whitelisted column list, not user input; values are prepared.
144	return (int) $wpdb->get_var($wpdb->prepare($sql, $values));
145	}
146
147	public function updateJob($jobData, $where = array())
148	{
149	global $wpdb;
150
151	if (empty($where) \|\| empty($jobData)) return false;
152
153	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Custom plugin table; no object cache for these operational queries.
154	return $wpdb->update($this->jobManagerTableName, $jobData, $where);
155	}
156
157	public function deleteJob($where = array())
158	{
159	global $wpdb;
160
161	if (empty($where)) return false;
162
163	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Custom plugin table; no object cache for these operational queries.
164	return $wpdb->delete($this->jobManagerTableName, $where);
165	}
166
167	public function ConvertStaleProcessingJobs($timeoutSeconds = 120)
168	{
169	global $wpdb;
170
171	$timeoutTimestamp = time() - $timeoutSeconds;
172
173	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, PluginCheck.Security.DirectDB.UnescapedDBParameter -- Custom plugin table; identifier from $wpdb->prefix, not user input.
174	$wpdb->query(
175	$wpdb->prepare(
176	"UPDATE {$this->jobManagerTableName}
177	SET status = CASE
178	WHEN attempts + 1 >= max_attempts THEN 'failed'
179	ELSE 'pending'
180	END,
181	attempts = attempts + 1,
182	started_at = NULL,
183	failed_at = CASE
184	WHEN attempts + 1 >= max_attempts THEN %d
185	ELSE failed_at
186	END
187	WHERE status = 'processing'
188	AND started_at IS NOT NULL
189	AND started_at < %d",
190	time(),
191	$timeoutTimestamp
192	)
193	);
194	}
195
196	public function hasProductJobInProgress(int $productId, string $jobType): bool
197	{
198	global $wpdb;
199
200	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, PluginCheck.Security.DirectDB.UnescapedDBParameter -- Custom plugin table; identifier from $wpdb->prefix, not user input.
201	$jobs = $wpdb->get_results($wpdb->prepare(
202	"SELECT payload FROM {$this->jobManagerTableName}
203	WHERE job_type = %s
204	AND (status = %s OR status = %s)",
205	$jobType,
206	'pending',
207	'processing'
208	));
209
210	if (empty($jobs)) {
211	return false;
212	}
213
214	foreach ($jobs as $job) {
215	$payload = json_decode($job->payload, true);
216	$jobProductId = $payload['product_id'] ?? $payload;
217
218	if (intval($jobProductId) === intval($productId)) {
219	return true;
220	}
221	}
222
223	return false;
224	}
225
226	public function retryJob(int $jobId, ?string $errorMessage = null): bool
227	{
228	global $wpdb;
229
230	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, PluginCheck.Security.DirectDB.UnescapedDBParameter -- Custom plugin table; identifier from $wpdb->prefix, not user input.
231	$job = $wpdb->get_row($wpdb->prepare(
232	"SELECT * FROM {$this->jobManagerTableName} WHERE id = %d",
233	$jobId
234	));
235
236	if (!$job) return false;
237
238	$newAttempts = intval($job->attempts) + 1;
239
240	$errorMessages = [];
241	if (!empty($job->error_message)) {
242	$decoded = json_decode($job->error_message, true);
243	if (json_last_error() === JSON_ERROR_NONE && is_array($decoded)) $errorMessages = $decoded;
244	}
245
246	if ($errorMessage) $errorMessages[$newAttempts] = $errorMessage;
247
248	$encodedErrors = json_encode($errorMessages, JSON_UNESCAPED_UNICODE);
249
250	if ($newAttempts >= intval($job->max_attempts)) {
251	$this->updateJob(
252	[
253	'status' => 'failed',
254	'error_message' => $encodedErrors,
255	'failed_at' => time(),
256	'started_at' => 0,
257	'attempts' => $newAttempts,
258	],
259	['id' => $jobId]
260	);
261	return false;
262	}
263
264	// Progressive exponential backoff: 30s, 60s, 120s, 240s, ...
265	$delaySeconds = 30 * (int) pow(2, $newAttempts - 1);
266	$retryAfter = time() + $delaySeconds;
267
268	// Atomic DELETE + INSERT inside a transaction so a crash can't lose the job.
269	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Transaction control for atomic job requeue; no object cache applicable.
270	$wpdb->query('START TRANSACTION');
271	try {
272	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Custom plugin table; no object cache for these operational queries.
273	$wpdb->delete($this->jobManagerTableName, ['id' => $jobId]);
274
275	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Custom plugin table; no object cache for these operational queries.
276	$wpdb->insert(
277	$this->jobManagerTableName,
278	[
279	'job_type' => $job->job_type,
280	'status' => 'pending',
281	'payload' => $job->payload,
282	'attempts' => $newAttempts,
283	'max_attempts' => $job->max_attempts,
284	'error_message' => $encodedErrors,
285	'created_at' => $job->created_at,
286	'retry_after' => $retryAfter,
287	'started_at' => 0,
288	]
289	);
290
291	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Transaction control for atomic job requeue; no object cache applicable.
292	$wpdb->query('COMMIT');
293	} catch (\Exception $e) {
294	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Transaction control for atomic job requeue; no object cache applicable.
295	$wpdb->query('ROLLBACK');
296	throw $e;
297	}
298
299	return true;
300	}
301
302	public function failJob(int $jobId, ?string $errorMessage = null): bool
303	{
304	return $this->updateJob(
305	[
306	'status' => 'failed',
307	'error_message' => $errorMessage,
308	'failed_at' => time(),
309	'started_at' => 0,
310	],
311	['id' => $jobId]
312	);
313	}
314
315	}
316