← All changes | JobManager.php +78 -13 1.9.1 → 1.10.22 View file →

		@@ -23,25 +23,33 @@
23	23	public function createJob($jobType, $status = 'pending', $payload = null, $maxAttempts = 3)
24	24	{
25	25	global $wpdb;
26	26
27		- return $wpdb->insert(
28		- $this->jobManagerTableName,
29		- array(
30		- 'job_type' => $jobType,
31		- 'status' => $status,
	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,
32	33	'payload' => $payload,
33	34	'attempts' => 0,
34	35	'max_attempts' => $maxAttempts,
35		- 'created_at' => time(),
36		- )
37		- );
38		- }
	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	+ }
39	46
40	47	public function getNextEligibleJob(string $jobType): ?object
41	48	{
42	49	global $wpdb;
43	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.
44	52	return $wpdb->get_row($wpdb->prepare(
45	53	"SELECT * FROM {$this->jobManagerTableName}
46	54	WHERE job_type = %s
47	55	AND status = 'pending'
		@@ -56,10 +64,55 @@
56	64	public function hasAnyProcessingJob(): bool
57	65	{
58	66	return $this->getCountJobs(['status' => 'processing']) > 0;
59	67	}
60		-
61		- public function getJob($where = array())
	68	+
	69	+ public function getProductUpdateStatus(): array
	70	+ {
	71	+ $quickJob = $this->getJob(['job_type' => 'sync_basalam_bulk_update_products', 'status' => 'processing'])
	72	+ ?: $this->getJob(['job_type' => 'sync_basalam_bulk_update_products', 'status' => 'pending']);
	73	+ $fullJob = $this->getJob(['job_type' => 'sync_basalam_update_all_products', 'status' => 'processing'])
	74	+ ?: $this->getJob(['job_type' => 'sync_basalam_update_all_products', 'status' => 'pending']);
	75	+ $singleCount = $this->getCountJobs([
	76	+ 'job_type' => 'sync_basalam_update_single_product',
	77	+ 'status' => ['pending', 'processing'],
	78	+ ]);
	79	+
	80	+ return [
	81	+ 'active' => (bool) ($quickJob || $fullJob || $singleCount > 0),
	82	+ 'type' => $quickJob ? 'quick' : (($fullJob || $singleCount > 0) ? 'full' : ''),
	83	+ 'status' => $quickJob ? $quickJob->status : ($fullJob ? $fullJob->status : ''),
	84	+ 'count' => $singleCount,
	85	+ ];
	86	+ }
	87	+
	88	+ public function hasPendingOrStaleProcessingJobs(int $staleProcessingTimeoutSeconds = 120): bool
	89	+ {
	90	+ global $wpdb;
	91	+
	92	+ $now = time();
	93	+ $staleBefore = $now - $staleProcessingTimeoutSeconds;
	94	+
	95	+ // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, PluginCheck.Security.DirectDB.UnescapedDBParameter -- Custom plugin table; identifier from $wpdb->prefix, not user input.
	96	+ $result = $wpdb->get_var($wpdb->prepare(
	97	+ "SELECT 1 FROM {$this->jobManagerTableName}
	98	+ WHERE (
	99	+ status = 'pending'
	100	+ AND (retry_after IS NULL OR retry_after <= %d)
	101	+ ) OR (
	102	+ status = 'processing'
	103	+ AND started_at IS NOT NULL
	104	+ AND started_at < %d
	105	+ )
	106	+ LIMIT 1",
	107	+ $now,
	108	+ $staleBefore
	109	+ ));
	110	+
	111	+ return (string) $result === '1';
	112	+ }
	113	+
	114	+ public function getJob($where = array())
62	115	{
63	116	global $wpdb;
64	117
65	118	if (empty($where)) return null;
		@@ -68,9 +121,9 @@
68	121	$values = [];
69	122
70	123	foreach ($where as $column => $value) {
71	124	if (!in_array($column, self::ALLOWED_COLUMNS, true)) {
72		- throw new \InvalidArgumentException("Invalid column: {$column}");
	125	+ throw new \InvalidArgumentException(esc_html("Invalid column: {$column}"));
73	126	}
74	127	$conditions[] = "{$column} = %s";
75	128	$values[] = $value;
76	129	}
		@@ -76,8 +129,9 @@
76	129	}
77	130
78	131	$sql = "SELECT * FROM {$this->jobManagerTableName} WHERE " . implode(" AND ", $conditions) . " LIMIT 1";
79	132
	133	+ // 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.
80	134	return $wpdb->get_row($wpdb->prepare($sql, $values));
81	135	}
82	136
83	137	public function getCountJobs($where = array())
		@@ -90,9 +144,9 @@
90	144	$values = [];
91	145
92	146	foreach ($where as $column => $value) {
93	147	if (!in_array($column, self::ALLOWED_COLUMNS, true)) {
94		- throw new \InvalidArgumentException("Invalid column: {$column}");
	148	+ throw new \InvalidArgumentException(esc_html("Invalid column: {$column}"));
95	149	}
96	150	if (is_array($value)) {
97	151	$placeholders = array_fill(0, count($value), '%s');
98	152	$conditions[] = "{$column} IN (" . implode(',', $placeholders) . ")";
		@@ -104,8 +158,9 @@
104	158	}
105	159
106	160	$sql = "SELECT COUNT(*) FROM {$this->jobManagerTableName} WHERE " . implode(" AND ", $conditions);
107	161
	162	+ // 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.
108	163	return (int) $wpdb->get_var($wpdb->prepare($sql, $values));
109	164	}
110	165
111	166	public function updateJob($jobData, $where = array())
		@@ -113,8 +168,9 @@
113	168	global $wpdb;
114	169
115	170	if (empty($where) || empty($jobData)) return false;
116	171
	172	+ // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Custom plugin table; no object cache for these operational queries.
117	173	return $wpdb->update($this->jobManagerTableName, $jobData, $where);
118	174	}
119	175
120	176	public function deleteJob($where = array())
		@@ -122,8 +178,9 @@
122	178	global $wpdb;
123	179
124	180	if (empty($where)) return false;
125	181
	182	+ // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Custom plugin table; no object cache for these operational queries.
126	183	return $wpdb->delete($this->jobManagerTableName, $where);
127	184	}
128	185
129	186	public function ConvertStaleProcessingJobs($timeoutSeconds = 120)
		@@ -131,8 +188,9 @@
131	188	global $wpdb;
132	189
133	190	$timeoutTimestamp = time() - $timeoutSeconds;
134	191
	192	+ // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, PluginCheck.Security.DirectDB.UnescapedDBParameter -- Custom plugin table; identifier from $wpdb->prefix, not user input.
135	193	$wpdb->query(
136	194	$wpdb->prepare(
137	195	"UPDATE {$this->jobManagerTableName}
138	196	SET status = CASE
		@@ -157,8 +215,9 @@
157	215	public function hasProductJobInProgress(int $productId, string $jobType): bool
158	216	{
159	217	global $wpdb;
160	218
	219	+ // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, PluginCheck.Security.DirectDB.UnescapedDBParameter -- Custom plugin table; identifier from $wpdb->prefix, not user input.
161	220	$jobs = $wpdb->get_results($wpdb->prepare(
162	221	"SELECT payload FROM {$this->jobManagerTableName}
163	222	WHERE job_type = %s
164	223	AND (status = %s OR status = %s)",
		@@ -186,8 +245,9 @@
186	245	public function retryJob(int $jobId, ?string $errorMessage = null): bool
187	246	{
188	247	global $wpdb;
189	248
	249	+ // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, PluginCheck.Security.DirectDB.UnescapedDBParameter -- Custom plugin table; identifier from $wpdb->prefix, not user input.
190	250	$job = $wpdb->get_row($wpdb->prepare(
191	251	"SELECT * FROM {$this->jobManagerTableName} WHERE id = %d",
192	252	$jobId
193	253	));
		@@ -224,12 +284,15 @@
224	284	$delaySeconds = 30 * (int) pow(2, $newAttempts - 1);
225	285	$retryAfter = time() + $delaySeconds;
226	286
227	287	// Atomic DELETE + INSERT inside a transaction so a crash can't lose the job.
	288	+ // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Transaction control for atomic job requeue; no object cache applicable.
228	289	$wpdb->query('START TRANSACTION');
229	290	try {
	291	+ // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Custom plugin table; no object cache for these operational queries.
230	292	$wpdb->delete($this->jobManagerTableName, ['id' => $jobId]);
231	293
	294	+ // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Custom plugin table; no object cache for these operational queries.
232	295	$wpdb->insert(
233	296	$this->jobManagerTableName,
234	297	[
235	298	'job_type' => $job->job_type,
		@@ -243,10 +306,12 @@
243	306	'started_at' => 0,
244	307	]
245	308	);
246	309
	310	+ // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Transaction control for atomic job requeue; no object cache applicable.
247	311	$wpdb->query('COMMIT');
248	312	} catch (\Exception $e) {
	313	+ // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Transaction control for atomic job requeue; no object cache applicable.
249	314	$wpdb->query('ROLLBACK');
250	315	throw $e;
251	316	}
252	317