sync-basalam / JobManager.php

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

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