wp-stateless / lib / classes / sync / class-background-sync.php

class-background-sync.php in WP-Stateless – Google Cloud Storage 3.2.1, at lib/classes/sync/class-background-sync.php

473 lines 12.3 KB

Raw Download Zip

1	<?php
2
3	namespace wpCloud\StatelessMedia\Sync;
4
5	// Require lib classes if not yet available
6	if (!class_exists('UDX_WP_Async_Request')) {
7	require_once ud_get_stateless_media()->path('lib/ns-vendor/classes/deliciousbrains/wp-background-processing/classes/wp-async-request.php', 'dir');
8	}
9
10	if (!class_exists('UDX_WP_Background_Process')) {
11	require_once ud_get_stateless_media()->path('lib/ns-vendor/classes/deliciousbrains/wp-background-processing/classes/wp-background-process.php', 'dir');
12	}
13
14	use UDX_WP_Background_Process, JsonSerializable;
15
16	/**
17	* Generic background process
18	*/
19	abstract class BackgroundSync extends UDX_WP_Background_Process implements ISync, JsonSerializable {
20
21	/**
22	* Cron Healthcheck interval
23	*/
24	public $cron_interval;
25
26	/**
27	* Flag to allow sorting
28	*/
29	protected $allow_sorting = false;
30
31	/**
32	* Flag to allow setting the limit
33	*/
34	protected $allow_limit = false;
35
36	/**
37	* Storage for emergency memory
38	*/
39	private $emergency_memory = null;
40
41	/**
42	* Storage for currently processed item
43	*/
44	protected $currently_processing_item = null;
45
46	/**
47	* Extend the construct
48	*/
49	public function __construct() {
50	// Support different threads for multisite installations
51	$blog_id = get_current_blog_id();
52	$this->action = "{$this->action}_{$blog_id}";
53
54	add_filter('wp_stateless_sync_types', function ($classes) {
55	$classes[$c = get_called_class()] = $c;
56	return $classes;
57	});
58
59	$this->cron_interval = $this->get_healthcheck_cron_interval();
60
61	// Reserve 1MB of RAM for the fallback action
62	$this->emergency_memory = new \SplFixedArray(65536);
63
64	// Register the fallback action to be executed on shutdown
65	register_shutdown_function(function () {
66	// Free up reserved memory
67	$this->emergency_memory = null;
68
69	// Check if we should execute the fallback action
70	if (is_null($err = error_get_last())) return;
71	if ($err['type'] != E_ERROR) return;
72	if (strstr($err['message'], 'memory') === false \|\| strstr($err['message'], 'exhausted') === false) return;
73	if (!$this->is_running()) return;
74	if (!$this->currently_processing_item) return;
75
76	// If we are here, then we shutdown because of `memory exhausted` error
77
78	// Remove already processed and problem items from the current batch
79	$current_batch = $this->get_batch();
80	if ($current_batch && $current_batch->data && is_array($current_batch->data)) {
81	foreach ($current_batch->data as $key => $item) {
82	unset($current_batch->data[$key]);
83	if ($item == $this->currently_processing_item) {
84	$this->log(sprintf(__('Item skipped: %s. Waiting for process to resume.', ud_get_stateless_media()->domain), $this->currently_processing_item));
85	call_user_func([get_class(), 'task'], $this->currently_processing_item);
86	break;
87	}
88	}
89	$current_batch->data = array_values($current_batch->data);
90	}
91
92	// Update current batch directly to the option
93	// because it needs to be updated even if it is empty
94	update_site_option($current_batch->key, $current_batch->data);
95
96	// Add notice
97	$this->save_process_meta([
98	'notice' => sprintf(
99	__("Not enough memory to process the following item '%s' %s: %s. Item skipped. Please, try to increase memory limit or use uploading by chunks: <a target=\"_blank\" href=\"https://wp-stateless.github.io/docs/constants/#wp_stateless_media_upload_chunk_size\">How to use WP_STATELESS_MEDIA_UPLOAD_CHUNK_SIZE setting.</a>", ud_get_stateless_media()->domain),
100	is_numeric($this->currently_processing_item) ? get_the_title($this->currently_processing_item) : $this->currently_processing_item,
101	is_numeric($this->currently_processing_item) ? "(ID: {$this->currently_processing_item})" : '',
102	$err['message']
103	)
104	]);
105
106	wp_die();
107	});
108
109	parent::__construct();
110	}
111
112	/**
113	* Maybe process queue (extended)
114	*
115	* Checks whether data exists within the queue and that
116	* the process is not already running.
117	*/
118	public function maybe_handle() {
119	// Don't lock up other requests while processing
120	session_write_close();
121
122	if ($this->is_process_running()) {
123	// Background process already running.
124	wp_die();
125	}
126
127	if ($this->is_queue_empty()) {
128	// No data to process.
129	wp_die();
130	}
131
132	$this->handle();
133
134	wp_die();
135	}
136
137	/**
138	* Determine sync healthcheck interval
139	*/
140	protected function get_healthcheck_cron_interval() {
141	return (defined('WP_STATELESS_SYNC_HEALTHCHECK_INTERVAL') && is_int(WP_STATELESS_SYNC_HEALTHCHECK_INTERVAL)) ? WP_STATELESS_SYNC_HEALTHCHECK_INTERVAL : 1;
142	}
143
144	/**
145	* Get option key for STOPPED option
146	*/
147	protected function get_stopped_option_key() {
148	return "{$this->action}_stopped";
149	}
150
151	/**
152	* Determine maximum batch size
153	*
154	* @return int Default is 50
155	*/
156	public function get_max_batch_size() {
157	return (defined('WP_STATELESS_SYNC_MAX_BATCH_SIZE') && is_int(WP_STATELESS_SYNC_MAX_BATCH_SIZE)) ? WP_STATELESS_SYNC_MAX_BATCH_SIZE : 50;
158	}
159
160	/**
161	* Get all batches
162	*
163	* @param int $limit 0
164	* @return array
165	*/
166	public function get_batches($limit = 0) {
167	global $wpdb;
168
169	if (empty($limit) \|\| !is_int($limit)) {
170	$limit = 0;
171	}
172
173	$table = $wpdb->options;
174	$column = 'option_name';
175	$key_column = 'option_id';
176	$value_column = 'option_value';
177
178	if (is_multisite()) {
179	$table = $wpdb->sitemeta;
180	$column = 'meta_key';
181	$key_column = 'meta_id';
182	$value_column = 'meta_value';
183	}
184
185	$key = $wpdb->esc_like($this->identifier) . '_batch_%';
186
187	$sql = "
188	SELECT *
189	FROM {$table}
190	WHERE {$column} LIKE %s
191	ORDER BY {$key_column} ASC
192	";
193
194	if (!empty($limit)) {
195	$sql .= " LIMIT {$limit}";
196	}
197
198	$items = $wpdb->get_results($wpdb->prepare($sql, $key));
199
200	$batches = [];
201
202	if (!empty($items)) {
203	$batches = array_map(
204	function ($item) use ($column, $value_column) {
205	$batch = new \stdClass();
206	$batch->key = $item->$column;
207	$batch->data = maybe_unserialize($item->$value_column);
208
209	return $batch;
210	},
211	$items
212	);
213	}
214
215	return $batches;
216	}
217
218	/**
219	* Get one top batch
220	*/
221	protected function get_batch() {
222	return array_reduce(
223	$this->get_batches(1),
224	function ($_, $batch) {
225	return $batch;
226	},
227	[]
228	);
229	}
230
231	/**
232	* Delete all batches
233	*
234	* @return self
235	*/
236	public function delete_all() {
237	$batches = $this->get_batches();
238
239	foreach ($batches as $batch) {
240	$this->delete($batch->key);
241	}
242
243	$this->clear_queue_size();
244	return $this;
245	}
246
247	/**
248	* Stop processing
249	*/
250	public function stop() {
251	$this->delete_all();
252	update_site_option($this->get_stopped_option_key(), true);
253	$this->clear_process_meta();
254	$this->log("Stopped");
255	}
256
257	/**
258	* Determine if process is stopped.
259	*
260	* @return bool
261	*/
262	public function is_stopped() {
263	$network_id = get_current_network_id();
264	wp_cache_delete("$network_id:notoptions", 'site-options');
265	return boolval(get_site_option($this->get_stopped_option_key()));
266	}
267
268	/**
269	* Update the whole queue size
270	*
271	* @param int $size
272	* @return self
273	*/
274	public function update_queue_size($size) {
275	$size = intval($size) + $this->get_queue_size();
276	update_site_option("{$this->action}_queue_size", $size);
277	return $this;
278	}
279
280	/**
281	* Get current queue size
282	*/
283	public function get_queue_size() {
284	return intval(get_site_option("{$this->action}_queue_size", 0));
285	}
286
287	/**
288	* Clear the queue size
289	*
290	* @return self
291	*/
292	public function clear_queue_size() {
293	delete_site_option("{$this->action}_queue_size");
294	return $this;
295	}
296
297	/**
298	* Clear process meta
299	*
300	* @return self
301	*/
302	public function clear_process_meta() {
303	// Clear limits for future starts
304	delete_site_option("{$this->action}_meta");
305	return $this;
306	}
307
308	/**
309	* Save process meta data
310	*
311	* @param array $meta
312	*/
313	public function save_process_meta($meta = []) {
314	if (!empty($meta)) {
315	$existing_meta = get_site_option("{$this->action}_meta", []);
316	foreach ($meta as $key => $value) {
317	$existing_meta[$key] = $value;
318	}
319	update_site_option("{$this->action}_meta", $existing_meta);
320	}
321	}
322
323	/**
324	* Get process meta data. All or by the key.
325	*
326	* @param string\|bool $key
327	* @return array\|string\|null
328	*/
329	public function get_process_meta($name = false) {
330	$meta = get_site_option("{$this->action}_meta", []);
331	if (false === $name) {
332	return $meta;
333	}
334	return isset($meta[$name]) ? $meta[$name] : null;
335	}
336
337	/**
338	* Extending save queue method
339	*
340	* @return $this
341	*/
342	public function save() {
343	$batch_size = is_array($this->data) ? count($this->data) : 1;
344	$this->update_queue_size($batch_size);
345	parent::save();
346	$this->data = [];
347	return $this;
348	}
349
350	/**
351	* Extending complete process method
352	*/
353	protected function complete() {
354	parent::complete();
355	$this->clear_process_meta();
356	$this->clear_queue_size();
357	delete_site_option($this->get_stopped_option_key());
358
359	if ($admin_email = get_option('admin_email')) {
360	$sync_name = strip_tags($this->get_name());
361	$site = site_url();
362	wp_mail(
363	$admin_email,
364	sprintf(__('Stateless Sync for %s is Complete', ud_get_stateless_media()->domain), $sync_name),
365	sprintf(__("This is a simple notification to inform you that the WP-Stateless plugin has finished a %s synchronization process for %s.\n\nIf you have WP_STATELESS_SYNC_LOG or WP_DEBUG_LOG enabled, check those logs to review any errors that may have occurred during the synchronization process.", ud_get_stateless_media()->domain), $sync_name, $site)
366	);
367	}
368	}
369
370	/**
371	* Remember currently processing item
372	*/
373	protected function before_task($item) {
374	$this->currently_processing_item = $item;
375	}
376
377	/**
378	* Common task that should be executed in the end of each subclass task
379	*/
380	protected function task($_) {
381	$processedCount = intval($this->get_process_meta('processed'));
382	$this->save_process_meta([
383	'processed' => ++$processedCount,
384	'last_at' => current_time('timestamp')
385	]);
386	}
387
388	/**
389	* Default name
390	*
391	* @return string
392	*/
393	public function get_name() {
394	return __('Background Sync', ud_get_stateless_media()->domain);
395	}
396
397	/**
398	* Default helper window is set to false
399	*
400	* @return HelperWindow\|bool
401	*/
402	public function get_helper_window() {
403	return false;
404	}
405
406	/**
407	* Process specific notice
408	*
409	* @return array\|bool
410	*/
411	public function get_process_notice() {
412	$notice = $this->get_process_meta('notice');
413	if (empty($notice)) return [];
414	return [$notice];
415	}
416
417	/**
418	* Is running?
419	*/
420	public function is_running() {
421	return !$this->is_queue_empty() \|\| $this->is_process_running();
422	}
423
424	/**
425	* Convert to json
426	*
427	* @return array
428	*/
429	public function jsonSerialize() {
430	return [
431	'id' => get_called_class(),
432	'name' => $this->get_name(),
433	'helper' => $this->get_helper_window(),
434	'is_running' => $this->is_running(),
435	'limit' => ($limit = $this->get_process_meta('limit')) ? $limit : 0,
436	'order' => ($order = $this->get_process_meta('order')) ? $order : 'desc',
437	'total_items' => $this->get_total_items(),
438	'queued_items' => $this->get_queue_size(),
439	'processed_items' => ($processed = $this->get_process_meta('processed')) ? $processed : 0,
440	'allow_limit' => $this->allow_limit,
441	'allow_sorting' => $this->allow_sorting,
442	'notice' => $this->get_process_notice()
443	];
444	}
445
446	/**
447	* Log background process event
448	*
449	* @param string $message
450	* @return bool TRUE on success or FALSE on failure
451	*/
452	public function log($message) {
453	$message = strip_tags(sprintf('Background Sync - %s: %s', $this->get_name(), $message));
454
455	if (is_multisite()) {
456	$blog_id = get_current_blog_id();
457	$message = sprintf('[Blog %s] %s', $blog_id, $message);
458	}
459
460	if (!defined('WP_STATELESS_SYNC_LOG')) {
461	return error_log($message);
462	}
463
464	return error_log(date('c') . ": $message\n", 3, WP_STATELESS_SYNC_LOG);
465	}
466
467	/**
468	* Start process.
469	* Should be implemented by subclasses.
470	*/
471	abstract public function start();
472	}
473