jetpack / json-endpoints / jetpack / class.jetpack-json-api-sync-endpoint.php

class.jetpack-json-api-sync-endpoint.php in Jetpack – WP Security, Backup, Speed, & Growth 12.9.5, at json-endpoints/jetpack/class.jetpack-json-api-sync-endpoint.php

636 lines 16.2 KB

Raw Download Zip

1	<?php // phpcs:ignore WordPress.Files.FileName.InvalidClassFileName
2
3	use Automattic\Jetpack\Sync\Actions;
4	use Automattic\Jetpack\Sync\Health;
5	use Automattic\Jetpack\Sync\Modules;
6	use Automattic\Jetpack\Sync\Queue;
7	use Automattic\Jetpack\Sync\Queue_Buffer;
8	use Automattic\Jetpack\Sync\Replicastore;
9	use Automattic\Jetpack\Sync\Sender;
10	use Automattic\Jetpack\Sync\Settings;
11
12	// phpcs:disable Generic.Files.OneObjectStructurePerFile.MultipleFound
13
14	/**
15	* Sync endpoint class.
16	*
17	* POST /sites/%s/sync
18	*/
19	class Jetpack_JSON_API_Sync_Endpoint extends Jetpack_JSON_API_Endpoint {
20
21	/**
22	* This endpoint allows authentication both via a blog and a user token.
23	* If a user token is used, that user should have `manage_options` capability.
24	*
25	* @var array\|string
26	*/
27	protected $needed_capabilities = 'manage_options';
28
29	/**
30	* Validate the call.
31	*
32	* @param int $_blog_id - the blog ID.
33	* @param string $capability - the capability.
34	* @param bool $check_manage_active - unused.
35	*
36	* @return bool\|WP_Error
37	*/
38	protected function validate_call( $_blog_id, $capability, $check_manage_active = true ) { // phpcs:ignore VariableAnalysis.CodeAnalysis.VariableAnalysis.UnusedVariable
39	return parent::validate_call( $_blog_id, $capability, false );
40	}
41
42	/**
43	* The result.
44	*
45	* @return array
46	*/
47	protected function result() {
48	$args = $this->input();
49	$modules = null;
50
51	// convert list of modules in comma-delimited format into an array
52	// of "$modulename => true"
53	if ( isset( $args['modules'] ) && ! empty( $args['modules'] ) ) {
54	$modules = array_map( '__return_true', array_flip( array_map( 'trim', explode( ',', $args['modules'] ) ) ) );
55	}
56
57	foreach ( array( 'posts', 'comments', 'users' ) as $module_name ) {
58	if ( 'users' === $module_name && isset( $args[ $module_name ] ) && 'initial' === $args[ $module_name ] ) {
59	$modules['users'] = 'initial';
60	} elseif ( isset( $args[ $module_name ] ) ) {
61	$ids = explode( ',', $args[ $module_name ] );
62	if ( is_countable( $ids ) && count( $ids ) > 0 ) {
63	$modules[ $module_name ] = $ids;
64	}
65	}
66	}
67
68	if ( empty( $modules ) ) {
69	$modules = null;
70	}
71	return array( 'scheduled' => Actions::do_full_sync( $modules ) );
72	}
73
74	/**
75	* Validate the queue.
76	*
77	* @param array $query - the query.
78	*
79	* @return string\|WP_Error
80	*/
81	protected function validate_queue( $query ) {
82	if ( ! isset( $query ) ) {
83	return new WP_Error( 'invalid_queue', 'Queue name is required', 400 );
84	}
85
86	if ( ! in_array( $query, array( 'sync', 'full_sync', 'immediate' ), true ) ) {
87	return new WP_Error( 'invalid_queue', 'Queue name should be sync, full_sync or immediate', 400 );
88	}
89	return $query;
90	}
91	}
92
93	/**
94	* Sync status endpoint class.
95	*
96	* GET /sites/%s/sync/status
97	*/
98	class Jetpack_JSON_API_Sync_Status_Endpoint extends Jetpack_JSON_API_Sync_Endpoint {
99	/**
100	* Callback for the endpoint.
101	*
102	* @return array
103	*/
104	protected function result() {
105	$args = $this->query_args();
106	$fields = isset( $args['fields'] ) ? $args['fields'] : array();
107	return Actions::get_sync_status( $fields );
108	}
109	}
110
111	/**
112	* Sync Check Endpoint class.
113	* GET /sites/%s/data-check
114	*/
115	class Jetpack_JSON_API_Sync_Check_Endpoint extends Jetpack_JSON_API_Sync_Endpoint {
116	/**
117	* Callback for the endpoint.
118	*
119	* @return array
120	*/
121	protected function result() {
122	Actions::mark_sync_read_only();
123	$store = new Replicastore();
124	return $store->checksum_all();
125	}
126	}
127
128	/**
129	* Sync histogram endpoint.
130	* GET /sites/%s/data-histogram
131	*/
132	class Jetpack_JSON_API_Sync_Histogram_Endpoint extends Jetpack_JSON_API_Sync_Endpoint {
133	/**
134	* Callback for endpoint.
135	*
136	* @return array
137	*/
138	protected function result() {
139	$args = $this->query_args();
140
141	if ( isset( $args['columns'] ) ) {
142	$columns = array_map( 'trim', explode( ',', $args['columns'] ) );
143	} else {
144	$columns = null; // go with defaults
145	}
146
147	$store = new Replicastore();
148
149	if ( ! isset( $args['strip_non_ascii'] ) ) {
150	$args['strip_non_ascii'] = true;
151	}
152
153	/**
154	* Hack: nullify the values of `start_id` and `end_id` if we're only requesting ranges.
155	*
156	* The endpoint doesn't support nullable values :(
157	*/
158	if ( true === $args['only_range_edges'] ) {
159	if ( 0 === $args['start_id'] ) {
160	$args['start_id'] = null;
161	}
162
163	if ( 0 === $args['end_id'] ) {
164	$args['end_id'] = null;
165	}
166	}
167
168	$histogram = $store->checksum_histogram( $args['object_type'], $args['buckets'], $args['start_id'], $args['end_id'], $columns, $args['strip_non_ascii'], $args['shared_salt'], $args['only_range_edges'], $args['detailed_drilldown'] );
169
170	// Hack to disable Sync during this call, so we can resolve faster.
171	Actions::mark_sync_read_only();
172
173	return array(
174	'histogram' => $histogram,
175	'type' => $store->get_checksum_type(),
176	);
177	}
178	}
179
180	/**
181	* POST /sites/%s/sync/health
182	*/
183	class Jetpack_JSON_API_Sync_Modify_Health_Endpoint extends Jetpack_JSON_API_Sync_Endpoint {
184
185	/**
186	* Callback for sync/health endpoint.
187	*
188	* @return array\|WP_Error result of request.
189	*/
190	protected function result() {
191	$args = $this->input();
192
193	switch ( $args['status'] ) {
194	case Health::STATUS_IN_SYNC:
195	case Health::STATUS_OUT_OF_SYNC:
196	Health::update_status( $args['status'] );
197	break;
198	default:
199	return new WP_Error( 'invalid_status', 'Invalid Sync Status Provided.' );
200	}
201
202	// re-fetch so we see what's really being stored.
203	return array(
204	'success' => Health::get_status(),
205	);
206	}
207	}
208
209	/**
210	* POST /sites/%s/sync/settings
211	*/
212	class Jetpack_JSON_API_Sync_Modify_Settings_Endpoint extends Jetpack_JSON_API_Sync_Endpoint {
213	/**
214	* The endpoint callback.
215	*
216	* @return array
217	*/
218	protected function result() {
219	$args = $this->input();
220
221	$sync_settings = Settings::get_settings();
222
223	foreach ( $args as $key => $value ) {
224	if ( $value !== false ) {
225	if ( is_numeric( $value ) ) {
226	$value = (int) $value;
227	}
228
229	// special case for sending empty arrays - a string with value 'empty'
230	if ( $value === 'empty' ) {
231	$value = array();
232	}
233
234	$sync_settings[ $key ] = $value;
235	}
236	}
237
238	Settings::update_settings( $sync_settings );
239
240	// re-fetch so we see what's really being stored
241	return Settings::get_settings();
242	}
243	}
244
245	/**
246	* GET /sites/%s/sync/settings
247	*/
248	class Jetpack_JSON_API_Sync_Get_Settings_Endpoint extends Jetpack_JSON_API_Sync_Endpoint {
249	/**
250	* Endpoint callback.
251	*
252	* @return array
253	*/
254	protected function result() {
255
256	return Settings::get_settings();
257	}
258	}
259
260	/**
261	* GET /sites/%s/sync/object
262	*/
263	class Jetpack_JSON_API_Sync_Object extends Jetpack_JSON_API_Sync_Endpoint {
264	/**
265	* Endpoint callback.
266	*
267	* @return array\|WP_Error
268	*/
269	protected function result() {
270	$args = $this->query_args();
271
272	$module_name = $args['module_name'];
273
274	$sync_module = Modules::get_module( $module_name );
275	if ( ! $sync_module ) {
276	return new WP_Error( 'invalid_module', 'You specified an invalid sync module' );
277	}
278
279	$object_type = $args['object_type'];
280	$object_ids = $args['object_ids'];
281
282	$codec = Sender::get_instance()->get_codec();
283
284	Actions::mark_sync_read_only();
285	Settings::set_is_syncing( true );
286	$objects = $codec->encode( $sync_module->get_objects_by_id( $object_type, $object_ids ) );
287	Settings::set_is_syncing( false );
288
289	return array(
290	'objects' => $objects,
291	'codec' => $codec->name(),
292	);
293	}
294	}
295
296	/**
297	* Sync Now endpoint class.
298	*/
299	class Jetpack_JSON_API_Sync_Now_Endpoint extends Jetpack_JSON_API_Sync_Endpoint {
300	/**
301	* Endpoint callback.
302	*
303	* @return array
304	*/
305	protected function result() {
306	$args = $this->input();
307	$queue_name = $this->validate_queue( $args['queue'] );
308
309	if ( is_wp_error( $queue_name ) ) {
310	return $queue_name;
311	}
312
313	$sender = Sender::get_instance();
314	$response = $sender->do_sync_for_queue( new Queue( $args['queue'] ) );
315
316	return array(
317	'response' => $response,
318	);
319	}
320	}
321
322	/**
323	* Sync checkout endpoint.
324	*/
325	class Jetpack_JSON_API_Sync_Checkout_Endpoint extends Jetpack_JSON_API_Sync_Endpoint {
326	/**
327	* Endpoint callback.
328	*
329	* @return array\|WP_Error
330	*/
331	protected function result() {
332	$args = $this->input();
333	$queue_name = $this->validate_queue( $args['queue'] );
334
335	if ( is_wp_error( $queue_name ) ) {
336	return $queue_name;
337	}
338
339	if ( $args['number_of_items'] < 1 \|\| $args['number_of_items'] > 100 ) {
340	return new WP_Error( 'invalid_number_of_items', 'Number of items needs to be an integer that is larger than 0 and less then 100', 400 );
341	}
342
343	$number_of_items = absint( $args['number_of_items'] );
344
345	if ( 'immediate' === $queue_name ) {
346	return $this->immediate_full_sync_pull( $number_of_items );
347	}
348
349	return $this->queue_pull( $queue_name, $number_of_items, $args );
350	}
351
352	/**
353	* Create a queue.
354	*
355	* @param string $queue_name - the queue name.
356	* @param int $number_of_items - the number of items.
357	* @param array $args - the arguments.
358	*
359	* @return array\|WP_Error
360	*/
361	public function queue_pull( $queue_name, $number_of_items, $args ) {
362	$queue = new Queue( $queue_name );
363
364	if ( 0 === $queue->size() ) {
365	return new WP_Error( 'queue_size', 'The queue is empty and there is nothing to send', 400 );
366	}
367
368	$sender = Sender::get_instance();
369
370	// try to give ourselves as much time as possible.
371	set_time_limit( 0 );
372
373	if ( $args['pop'] ) {
374	$buffer = new Queue_Buffer( 'pop', $queue->pop( $number_of_items ) );
375	} else {
376	// let's delete the checkin state.
377	if ( $args['force'] ) {
378	$queue->unlock();
379	}
380	$buffer = $this->get_buffer( $queue, $number_of_items );
381	}
382	// Check that the $buffer is not checkout out already.
383	if ( is_wp_error( $buffer ) ) {
384	return new WP_Error( 'buffer_open', "We couldn't get the buffer it is currently checked out", 400 );
385	}
386
387	if ( ! is_object( $buffer ) ) {
388	return new WP_Error( 'buffer_non-object', 'Buffer is not an object', 400 );
389	}
390
391	Settings::set_is_syncing( true );
392	list( $items_to_send, $skipped_items_ids ) = $sender->get_items_to_send( $buffer, $args['encode'] );
393	Settings::set_is_syncing( false );
394
395	return array(
396	'buffer_id' => $buffer->id,
397	'items' => $items_to_send,
398	'skipped_items' => $skipped_items_ids,
399	'codec' => $args['encode'] ? $sender->get_codec()->name() : null,
400	'sent_timestamp' => time(),
401	);
402	}
403
404	/**
405	* The items.
406	*
407	* @var array
408	*/
409	public $items = array();
410
411	/**
412	* Send the data listener.
413	*/
414	public function jetpack_sync_send_data_listener() {
415	foreach ( func_get_args()[0] as $key => $item ) {
416	$this->items[ $key ] = $item;
417	}
418	}
419
420	/**
421	* Check out a buffer of full sync actions.
422	*
423	* @param null $number_of_items Number of Actions to check-out.
424	*
425	* @return array Sync Actions to be returned to requestor
426	*/
427	public function immediate_full_sync_pull( $number_of_items = null ) { // phpcs:ignore VariableAnalysis.CodeAnalysis.VariableAnalysis.UnusedVariable
428	// try to give ourselves as much time as possible.
429	set_time_limit( 0 );
430
431	$original_send_data_cb = array( 'Automattic\Jetpack\Sync\Actions', 'send_data' );
432	$temp_send_data_cb = array( $this, 'jetpack_sync_send_data_listener' );
433
434	Sender::get_instance()->set_enqueue_wait_time( 0 );
435	remove_filter( 'jetpack_sync_send_data', $original_send_data_cb );
436	add_filter( 'jetpack_sync_send_data', $temp_send_data_cb, 10, 6 );
437	Sender::get_instance()->do_full_sync();
438	remove_filter( 'jetpack_sync_send_data', $temp_send_data_cb );
439	add_filter( 'jetpack_sync_send_data', $original_send_data_cb, 10, 6 );
440
441	return array(
442	'items' => $this->items,
443	'codec' => Sender::get_instance()->get_codec()->name(),
444	'sent_timestamp' => time(),
445	'status' => Actions::get_sync_status(),
446	);
447	}
448
449	/**
450	* Get the queue buffer.
451	*
452	* @param object $queue - the queue.
453	* @param int $number_of_items - the number of items.
454	*
455	* @return Automattic\Jetpack\Sync\Queue_Buffer\|bool\|int\|\WP_Error
456	*/
457	protected function get_buffer( $queue, $number_of_items ) {
458	$start = time();
459	$max_duration = 5; // this will try to get the buffer
460
461	$buffer = $queue->checkout( $number_of_items );
462	$duration = time() - $start;
463
464	while ( is_wp_error( $buffer ) && $duration < $max_duration ) {
465	sleep( 2 );
466	$duration = time() - $start;
467	$buffer = $queue->checkout( $number_of_items );
468	}
469
470	if ( $buffer === false ) {
471	return new WP_Error( 'queue_size', 'The queue is empty and there is nothing to send', 400 );
472	}
473
474	return $buffer;
475	}
476	}
477
478	/**
479	* Close endpoint class.
480	*/
481	class Jetpack_JSON_API_Sync_Close_Endpoint extends Jetpack_JSON_API_Sync_Endpoint {
482	/**
483	* Endpoint callback.
484	*
485	* @return array\|WP_Error
486	*/
487	protected function result() {
488
489	$request_body = $this->input();
490	$queue_name = $this->validate_queue( $request_body['queue'] );
491
492	if ( is_wp_error( $queue_name ) ) {
493	return $queue_name;
494	}
495
496	if ( ! isset( $request_body['buffer_id'] ) ) {
497	return new WP_Error( 'missing_buffer_id', 'Please provide a buffer id', 400 );
498	}
499
500	if ( ! isset( $request_body['item_ids'] ) \|\| ! is_array( $request_body['item_ids'] ) ) {
501	return new WP_Error( 'missing_item_ids', 'Please provide a list of item ids in the item_ids argument', 400 );
502	}
503
504	$request_body['buffer_id'] = preg_replace( '/[^A-Za-z0-9]/', '', $request_body['buffer_id'] );
505	$request_body['item_ids'] = array_filter( array_map( array( 'Jetpack_JSON_API_Sync_Close_Endpoint', 'sanitize_item_ids' ), $request_body['item_ids'] ) );
506
507	$queue = new Queue( $queue_name );
508
509	$items = $queue->peek_by_id( $request_body['item_ids'] );
510
511	// Update Full Sync Status if queue is "full_sync".
512	if ( 'full_sync' === $queue_name ) {
513	$full_sync_module = Modules::get_module( 'full-sync' );
514
515	$full_sync_module->update_sent_progress_action( $items );
516	}
517
518	$buffer = new Queue_Buffer( $request_body['buffer_id'], $request_body['item_ids'] );
519	$response = $queue->close( $buffer, $request_body['item_ids'] );
520
521	// Perform another checkout?
522	if ( isset( $request_body['continue'] ) && $request_body['continue'] ) {
523	if ( in_array( $queue_name, array( 'full_sync', 'immediate' ), true ) ) {
524	// Send Full Sync Actions.
525	Sender::get_instance()->do_full_sync();
526	} elseif ( $queue->has_any_items() ) {
527	// Send Incremental Sync Actions.
528	Sender::get_instance()->do_sync();
529	}
530	}
531
532	if ( is_wp_error( $response ) ) {
533	return $response;
534	}
535
536	return array(
537	'success' => $response,
538	'status' => Actions::get_sync_status(),
539	);
540	}
541
542	/**
543	* Sanitize item IDs.
544	*
545	* @param string $item - the item we're sanitizing.
546	*
547	* @return string\|null
548	*/
549	protected static function sanitize_item_ids( $item ) {
550	// lets not delete any options that don't start with jpsq_sync-
551	if ( ! is_string( $item ) \|\| ! str_starts_with( $item, 'jpsq_' ) ) {
552	return null;
553	}
554	// Limit to A-Z,a-z,0-9,_,-,.
555	return preg_replace( '/[^A-Za-z0-9-_.]/', '', $item );
556	}
557	}
558
559	/**
560	* Unlock ednpoint class.
561	*/
562	class Jetpack_JSON_API_Sync_Unlock_Endpoint extends Jetpack_JSON_API_Sync_Endpoint {
563	/**
564	* Endpoint callback.
565	*
566	* @return array\|WP_Error
567	*/
568	protected function result() {
569	$args = $this->input();
570
571	if ( ! isset( $args['queue'] ) ) {
572	return new WP_Error( 'invalid_queue', 'Queue name is required', 400 );
573	}
574
575	if ( ! in_array( $args['queue'], array( 'sync', 'full_sync' ), true ) ) {
576	return new WP_Error( 'invalid_queue', 'Queue name should be sync or full_sync', 400 );
577	}
578
579	$queue = new Queue( $args['queue'] );
580
581	// False means that there was no lock to delete.
582	$response = $queue->unlock();
583	return array(
584	'success' => $response,
585	);
586	}
587	}
588
589	/**
590	* Object ID range class.
591	*/
592	class Jetpack_JSON_API_Sync_Object_Id_Range extends Jetpack_JSON_API_Sync_Endpoint {
593	/**
594	* Endpoint callback.
595	*
596	* @return array\|WP_Error
597	*/
598	protected function result() {
599	$args = $this->query_args();
600
601	$module_name = $args['sync_module'];
602	$batch_size = $args['batch_size'];
603
604	if ( ! $this->is_valid_sync_module( $module_name ) ) {
605	return new WP_Error( 'invalid_module', 'This sync module cannot be used to calculate a range.', 400 );
606	}
607
608	$module = Modules::get_module( $module_name );
609
610	return array(
611	'ranges' => $module->get_min_max_object_ids_for_batches( $batch_size ),
612	);
613	}
614
615	/**
616	* Check if sync module is valid.
617	*
618	* @param string $module_name - the module name.
619	*
620	* @return bool
621	*/
622	protected function is_valid_sync_module( $module_name ) {
623	return in_array(
624	$module_name,
625	array(
626	'comments',
627	'posts',
628	'terms',
629	'term_relationships',
630	'users',
631	),
632	true
633	);
634	}
635	}
636