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

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

474 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=\"%s\">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	ud_get_stateless_media()->get_docs_page_url('docs/constants/#wpstatelessmediauploadchunksize'),
104	)
105	]);
106
107	wp_die();
108	});
109
110	parent::__construct();
111	}
112
113	/**
114	* Maybe process queue (extended)
115	*
116	* Checks whether data exists within the queue and that
117	* the process is not already running.
118	*/
119	public function maybe_handle() {
120	// Don't lock up other requests while processing
121	session_write_close();
122
123	if ($this->is_processing()) {
124	// Background process already running.
125	wp_die();
126	}
127
128	if ($this->is_queue_empty()) {
129	// No data to process.
130	wp_die();
131	}
132
133	$this->handle();
134
135	wp_die();
136	}
137
138	/**
139	* Determine sync healthcheck interval
140	*/
141	protected function get_healthcheck_cron_interval() {
142	return (defined('WP_STATELESS_SYNC_HEALTHCHECK_INTERVAL') && is_int(WP_STATELESS_SYNC_HEALTHCHECK_INTERVAL)) ? WP_STATELESS_SYNC_HEALTHCHECK_INTERVAL : 1;
143	}
144
145	/**
146	* Get option key for STOPPED option
147	*/
148	protected function get_stopped_option_key() {
149	return "{$this->action}_stopped";
150	}
151
152	/**
153	* Determine maximum batch size
154	*
155	* @return int Default is 50
156	*/
157	public function get_max_batch_size() {
158	return (defined('WP_STATELESS_SYNC_MAX_BATCH_SIZE') && is_int(WP_STATELESS_SYNC_MAX_BATCH_SIZE)) ? WP_STATELESS_SYNC_MAX_BATCH_SIZE : 50;
159	}
160
161	/**
162	* Get all batches
163	*
164	* @param int $limit 0
165	* @return array
166	*/
167	public function get_batches($limit = 0) {
168	global $wpdb;
169
170	if (empty($limit) \|\| !is_int($limit)) {
171	$limit = 0;
172	}
173
174	$table = $wpdb->options;
175	$column = 'option_name';
176	$key_column = 'option_id';
177	$value_column = 'option_value';
178
179	if (is_multisite()) {
180	$table = $wpdb->sitemeta;
181	$column = 'meta_key';
182	$key_column = 'meta_id';
183	$value_column = 'meta_value';
184	}
185
186	$key = $wpdb->esc_like($this->identifier) . '_batch_%';
187
188	$sql = "
189	SELECT *
190	FROM {$table}
191	WHERE {$column} LIKE %s
192	ORDER BY {$key_column} ASC
193	";
194
195	if (!empty($limit)) {
196	$sql .= " LIMIT {$limit}";
197	}
198
199	$items = $wpdb->get_results($wpdb->prepare($sql, $key));
200
201	$batches = [];
202
203	if (!empty($items)) {
204	$batches = array_map(
205	function ($item) use ($column, $value_column) {
206	$batch = new \stdClass();
207	$batch->key = $item->$column;
208	$batch->data = maybe_unserialize($item->$value_column);
209
210	return $batch;
211	},
212	$items
213	);
214	}
215
216	return $batches;
217	}
218
219	/**
220	* Get one top batch
221	*/
222	protected function get_batch() {
223	return array_reduce(
224	$this->get_batches(1),
225	function ($_, $batch) {
226	return $batch;
227	},
228	[]
229	);
230	}
231
232	/**
233	* Delete all batches
234	*
235	* @return self
236	*/
237	public function delete_all() {
238	$batches = $this->get_batches();
239
240	foreach ($batches as $batch) {
241	$this->delete($batch->key);
242	}
243
244	$this->clear_queue_size();
245	return $this;
246	}
247
248	/**
249	* Stop processing
250	*/
251	public function stop() {
252	$this->delete_all();
253	update_site_option($this->get_stopped_option_key(), true);
254	$this->clear_process_meta();
255	$this->log("Stopped");
256	}
257
258	/**
259	* Determine if process is stopped.
260	*
261	* @return bool
262	*/
263	public function is_stopped() {
264	$network_id = get_current_network_id();
265	wp_cache_delete("$network_id:notoptions", 'site-options');
266	return boolval(get_site_option($this->get_stopped_option_key()));
267	}
268
269	/**
270	* Update the whole queue size
271	*
272	* @param int $size
273	* @return self
274	*/
275	public function update_queue_size($size) {
276	$size = intval($size) + $this->get_queue_size();
277	update_site_option("{$this->action}_queue_size", $size);
278	return $this;
279	}
280
281	/**
282	* Get current queue size
283	*/
284	public function get_queue_size() {
285	return intval(get_site_option("{$this->action}_queue_size", 0));
286	}
287
288	/**
289	* Clear the queue size
290	*
291	* @return self
292	*/
293	public function clear_queue_size() {
294	delete_site_option("{$this->action}_queue_size");
295	return $this;
296	}
297
298	/**
299	* Clear process meta
300	*
301	* @return self
302	*/
303	public function clear_process_meta() {
304	// Clear limits for future starts
305	delete_site_option("{$this->action}_meta");
306	return $this;
307	}
308
309	/**
310	* Save process meta data
311	*
312	* @param array $meta
313	*/
314	public function save_process_meta($meta = []) {
315	if (!empty($meta)) {
316	$existing_meta = get_site_option("{$this->action}_meta", []);
317	foreach ($meta as $key => $value) {
318	$existing_meta[$key] = $value;
319	}
320	update_site_option("{$this->action}_meta", $existing_meta);
321	}
322	}
323
324	/**
325	* Get process meta data. All or by the key.
326	*
327	* @param string\|bool $key
328	* @return array\|string\|null
329	*/
330	public function get_process_meta($name = false) {
331	$meta = get_site_option("{$this->action}_meta", []);
332	if (false === $name) {
333	return $meta;
334	}
335	return isset($meta[$name]) ? $meta[$name] : null;
336	}
337
338	/**
339	* Extending save queue method
340	*
341	* @return $this
342	*/
343	public function save() {
344	$batch_size = is_array($this->data) ? count($this->data) : 1;
345	$this->update_queue_size($batch_size);
346	parent::save();
347	$this->data = [];
348	return $this;
349	}
350
351	/**
352	* Extending complete process method
353	*/
354	protected function complete() {
355	parent::complete();
356	$this->clear_process_meta();
357	$this->clear_queue_size();
358	delete_site_option($this->get_stopped_option_key());
359
360	// Sending notification
361	$sync_name = strip_tags($this->get_name());
362	$site = site_url();
363
364	$subject = sprintf(__('WP-Stateless: %s Synchronization Complete', ud_get_stateless_media()->domain), $sync_name);
365	$message = sprintf(__("WP-Stateless has finished synchronizing %s for %s.\n\nIf you have WP_STATELESS_SYNC_LOG or WP_DEBUG_LOG enabled, review those logs now to review any errors that may have occurred during the synchronization process.", ud_get_stateless_media()->domain), $sync_name, $site);
366
367	do_action('wp_stateless_send_admin_email', $subject, $message);
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_processing();
422	}
423
424	/**
425	* Convert to json
426	*
427	* @return array
428	*/
429	#[\ReturnTypeWillChange]
430	public function jsonSerialize() {
431	return [
432	'id' => get_called_class(),
433	'name' => $this->get_name(),
434	'helper' => $this->get_helper_window(),
435	'is_running' => $this->is_running(),
436	'limit' => ($limit = $this->get_process_meta('limit')) ? $limit : 0,
437	'order' => ($order = $this->get_process_meta('order')) ? $order : 'desc',
438	'total_items' => $this->get_total_items(),
439	'queued_items' => $this->get_queue_size(),
440	'processed_items' => ($processed = $this->get_process_meta('processed')) ? $processed : 0,
441	'allow_limit' => $this->allow_limit,
442	'allow_sorting' => $this->allow_sorting,
443	'notice' => $this->get_process_notice()
444	];
445	}
446
447	/**
448	* Log background process event
449	*
450	* @param string $message
451	* @return bool TRUE on success or FALSE on failure
452	*/
453	public function log($message) {
454	$message = strip_tags(sprintf('Background Sync - %s: %s', $this->get_name(), $message));
455
456	if (is_multisite()) {
457	$blog_id = get_current_blog_id();
458	$message = sprintf('[Blog %s] %s', $blog_id, $message);
459	}
460
461	if (!defined('WP_STATELESS_SYNC_LOG')) {
462	return error_log($message);
463	}
464
465	return error_log(date('c') . ": $message\n", 3, WP_STATELESS_SYNC_LOG);
466	}
467
468	/**
469	* Start process.
470	* Should be implemented by subclasses.
471	*/
472	abstract public function start();
473	}
474