sync-basalam / JobManager.php

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

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