jetpack / jetpack_vendor / automattic / jetpack-sync / src / class-sender.php

class-sender.php in Jetpack – WP Security, Backup, Speed, & Growth 16.3-a.7, at jetpack_vendor/automattic/jetpack-sync/src/class-sender.php

1,040 lines 28.9 KB

Raw Download Zip

1	<?php
2	/**
3	* Sync sender.
4	*
5	* @package automattic/jetpack-sync
6	*/
7
8	namespace Automattic\Jetpack\Sync;
9
10	use Automattic\Jetpack\Connection\Manager;
11	use Automattic\Jetpack\Constants;
12	use WP_Error;
13
14	/**
15	* This class grabs pending actions from the queue and sends them
16	*/
17	class Sender {
18	/**
19	* Name of the option that stores the time of the next sync.
20	*
21	* @access public
22	*
23	* @var string
24	*/
25	const NEXT_SYNC_TIME_OPTION_NAME = 'jetpack_next_sync_time';
26
27	/**
28	* Name of the transient responsible for temprorarily disabling Sync sending during Pulls.
29	*
30	* @access public
31	*
32	* @var string
33	*/
34	const TEMP_SYNC_DISABLE_TRANSIENT_NAME = 'jetpack_disable_sync_sending';
35
36	/**
37	* Expiry of the transient responsible for temprorarily disabling Sync sending during Pulls.
38	*
39	* @access public
40	*
41	* @var int
42	*/
43	const TEMP_SYNC_DISABLE_TRANSIENT_EXPIRY = MINUTE_IN_SECONDS;
44
45	/**
46	* Sync timeout after a WPCOM error.
47	*
48	* @access public
49	*
50	* @var int
51	*/
52	const WPCOM_ERROR_SYNC_DELAY = 60;
53
54	/**
55	* Sync timeout after a queue has been locked.
56	*
57	* @access public
58	*
59	* @var int
60	*/
61	const QUEUE_LOCKED_SYNC_DELAY = 10;
62
63	/**
64	* Maximum bytes to checkout without exceeding the memory limit.
65	*
66	* @access private
67	*
68	* @var int
69	*/
70	private $dequeue_max_bytes;
71
72	/**
73	* Maximum bytes in a single encoded item.
74	*
75	* @access private
76	*
77	* @var int
78	*/
79	private $upload_max_bytes;
80
81	/**
82	* Maximum number of sync items in a single action.
83	*
84	* @access private
85	*
86	* @var int
87	*/
88	private $upload_max_rows;
89
90	/**
91	* Maximum time for perfirming a checkout of items from the queue (in seconds).
92	*
93	* @access private
94	*
95	* @var int
96	*/
97	private $max_dequeue_time;
98
99	/**
100	* How many seconds to wait after sending sync items after exceeding the sync wait threshold (in seconds).
101	*
102	* @access private
103	*
104	* @var int
105	*/
106	private $sync_wait_time;
107
108	/**
109	* How much maximum time to wait for the checkout to finish (in seconds).
110	*
111	* @access private
112	*
113	* @var int
114	*/
115	private $sync_wait_threshold;
116
117	/**
118	* How much maximum time to wait for the sync items to be queued for sending (in seconds).
119	*
120	* @access private
121	*
122	* @var int
123	*/
124	private $enqueue_wait_time;
125
126	/**
127	* Incremental sync queue object.
128	*
129	* @access private
130	*
131	* @var \Automattic\Jetpack\Sync\Queue
132	*/
133	private $sync_queue;
134
135	/**
136	* Full sync queue object.
137	*
138	* @access private
139	*
140	* @var \Automattic\Jetpack\Sync\Queue
141	*/
142	private $full_sync_queue;
143
144	/**
145	* Codec object for encoding and decoding sync items.
146	*
147	* @access private
148	*
149	* @var \Automattic\Jetpack\Sync\Codec_Interface
150	*/
151	private $codec;
152
153	/**
154	* The current user before we change or clear it.
155	*
156	* @access private
157	*
158	* @var \WP_User
159	*/
160	private $old_user;
161
162	/**
163	* Container for the singleton instance of this class.
164	*
165	* @access private
166	* @static
167	*
168	* @var \Automattic\Jetpack\Sync\Sender
169	*/
170	private static $instance;
171
172	/**
173	* Retrieve the singleton instance of this class.
174	*
175	* @access public
176	* @static
177	*
178	* @return Sender
179	*/
180	public static function get_instance() {
181	if ( null === self::$instance ) {
182	self::$instance = new self();
183	}
184
185	return self::$instance;
186	}
187
188	/**
189	* Constructor.
190	* This is necessary because you can't use "new" when you declare instance properties >:(
191	*
192	* @access protected
193	* @static
194	*/
195	protected function __construct() {
196	$this->set_defaults();
197	$this->init();
198	}
199
200	/**
201	* Initialize the sender.
202	* Prepares the current user and initializes all sync modules.
203	*
204	* @access private
205	*/
206	private function init() {
207	add_action( 'jetpack_sync_before_send_queue_sync', array( $this, 'maybe_set_user_from_token' ), 1 );
208	add_action( 'jetpack_sync_before_send_queue_sync', array( $this, 'maybe_clear_user_from_token' ), 20 );
209	add_filter( 'jetpack_xmlrpc_unauthenticated_methods', array( $this, 'register_jetpack_xmlrpc_methods' ) );
210	foreach ( Modules::get_modules() as $module ) {
211	$module->init_before_send();
212	}
213	}
214
215	/**
216	* Detect if this is a XMLRPC request with a valid signature.
217	* If so, changes the user to the new one.
218	*
219	* @access public
220	*/
221	public function maybe_set_user_from_token() {
222	$connection = new Manager();
223	$verified_user = $connection->verify_xml_rpc_signature();
224	if ( Constants::is_true( 'XMLRPC_REQUEST' ) &&
225	! is_wp_error( $verified_user )
226	&& $verified_user
227	) {
228	$old_user = wp_get_current_user();
229	$this->old_user = $old_user->ID ?? 0;
230	wp_set_current_user( $verified_user['user_id'] );
231	}
232	}
233
234	/**
235	* If we used to have a previous current user, revert back to it.
236	*
237	* @access public
238	*/
239	public function maybe_clear_user_from_token() {
240	if ( isset( $this->old_user ) ) {
241	wp_set_current_user( $this->old_user );
242	}
243	}
244
245	/**
246	* Retrieve the next sync time.
247	*
248	* Update @since 1.43.2
249	* Sometimes when we process Sync requests in Jetpack, the server clock can be a
250	* bit in the future and this can lock Sync to not send stuff for a while.
251	* We are introducing an extra check, to make sure to limit the next_sync_time
252	* to be at most one hour in the future from the current time.
253	*
254	* @access public
255	*
256	* @param string $queue_name Name of the queue.
257	* @return float Timestamp of the next sync.
258	*/
259	public function get_next_sync_time( $queue_name ) {
260	$option_name = self::NEXT_SYNC_TIME_OPTION_NAME . '_' . $queue_name;
261	$next_sync_time = (float) get_option( $option_name, 0 );
262
263	$is_more_than_one_hour = ( $next_sync_time - microtime( true ) ) >= HOUR_IN_SECONDS;
264
265	if ( $is_more_than_one_hour ) {
266	delete_option( $option_name );
267	$next_sync_time = 0;
268	}
269
270	return $next_sync_time;
271	}
272
273	/**
274	* Set the next sync time.
275	*
276	* @access public
277	*
278	* @param int $time Timestamp of the next sync.
279	* @param string $queue_name Name of the queue.
280	* @return boolean True if update was successful, false otherwise.
281	*/
282	public function set_next_sync_time( $time, $queue_name ) {
283	return update_option( self::NEXT_SYNC_TIME_OPTION_NAME . '_' . $queue_name, $time, true );
284	}
285
286	/**
287	* Trigger a full sync.
288	*
289	* @access public
290	*
291	* @return boolean\|WP_Error True if this sync sending was successful, error object otherwise.
292	*/
293	public function do_full_sync() {
294	$sync_module = Modules::get_module( 'full-sync' );
295	'@phan-var Modules\Full_Sync_Immediately\|Modules\Full_Sync $sync_module';
296	if ( ! $sync_module ) {
297	return;
298	}
299	// Full Sync Disabled.
300	if ( ! Settings::get_setting( 'full_sync_sender_enabled' ) ) {
301	return;
302	}
303
304	// Don't sync if request is marked as read only.
305	if ( Constants::is_true( 'JETPACK_SYNC_READ_ONLY' ) ) {
306	return new WP_Error( 'jetpack_sync_read_only' );
307	}
308
309	// Sync not started or Sync finished.
310	$status = $sync_module->get_status();
311	if ( false === $status['started'] \|\| ( ! empty( $status['started'] ) && ! empty( $status['finished'] ) ) ) {
312	return false;
313	}
314
315	$this->continue_full_sync_enqueue();
316	// immediate full sync sends data in continue_full_sync_enqueue.
317	if ( ! $sync_module instanceof Modules\Full_Sync_Immediately ) {
318	return $this->do_sync_and_set_delays( $this->full_sync_queue );
319	} else {
320	$status = $sync_module->get_status();
321	// Sync not started or Sync finished.
322	if ( false === $status['started'] \|\| ( ! empty( $status['started'] ) && ! empty( $status['finished'] ) ) ) {
323	return false;
324	} else {
325	return true;
326	}
327	}
328	}
329
330	/**
331	* Enqueue the next sync items for sending.
332	* Will not be done if the current request is a WP import one.
333	* Will be delayed until the next sync time comes.
334	*
335	* @access private
336	*/
337	private function continue_full_sync_enqueue() {
338	if ( defined( 'WP_IMPORTING' ) && WP_IMPORTING ) {
339	return false;
340	}
341
342	if ( $this->get_next_sync_time( 'full-sync-enqueue' ) > microtime( true ) ) {
343	return false;
344	}
345
346	$full_sync_module = Modules::get_module( 'full-sync' );
347	'@phan-var Modules\Full_Sync_Immediately\|Modules\Full_Sync $full_sync_module';
348	$full_sync_module->continue_enqueuing();
349
350	$this->set_next_sync_time( time() + $this->get_enqueue_wait_time(), 'full-sync-enqueue' );
351	}
352
353	/**
354	* Trigger incremental sync.
355	*
356	* @access public
357	*
358	* @return boolean\|WP_Error True if this sync sending was successful, error object otherwise.
359	*/
360	public function do_sync() {
361	// Sync directly during cron. We are doing this because otherwise
362	// the dedicated sync flow would be spawning HTTP requests during cron shutdown,
363	// which can be unreliable and cause sync lag for time-sensitive events like updates.
364	if ( ! Settings::is_dedicated_sync_enabled() \|\| Settings::is_doing_cron() ) {
365	$result = $this->do_sync_and_set_delays( $this->sync_queue );
366	} else {
367	$result = Dedicated_Sender::spawn_sync( $this->sync_queue );
368	}
369
370	return $result;
371	}
372
373	/**
374	* Trigger incremental sync and early exit on Dedicated Sync request.
375	*
376	* @access public
377	*
378	* @param bool $do_real_exit If we should exit at the end of the request. We should by default.
379	* In the context of running this in the REST API, we actually want to return an error.
380	*
381	* @return void\|WP_Error
382	*/
383	public function do_dedicated_sync_and_exit( $do_real_exit = true ) {
384	nocache_headers();
385
386	if ( ! Settings::is_dedicated_sync_enabled() ) {
387	return new WP_Error( 'dedicated_sync_disabled', 'Dedicated Sync flow is disabled.' );
388	}
389
390	if ( ! Dedicated_Sender::is_dedicated_sync_request() ) {
391	return new WP_Error( 'non_dedicated_sync_request', 'Not a Dedicated Sync request.' );
392	}
393
394	/**
395	* Output an `OK` to show that Dedicated Sync is enabled and we can process events.
396	* This is used to test the feature is working.
397	*
398	* @see \Automattic\Jetpack\Sync\Dedicated_Sender::can_spawn_dedicated_sync_request
399	*/
400	// phpcs:ignore WordPress.Security.EscapeOutput.OutputNotEscaped -- This is just a constant string used for Validation.
401	echo Dedicated_Sender::DEDICATED_SYNC_VALIDATION_STRING;
402
403	// Try to disconnect the request as quickly as possible and process things in the background.
404	$this->fastcgi_finish_request();
405
406	/**
407	* Close the PHP session to free up the server threads to handle other requests while we
408	* send sync data with Dedicated Sync.
409	*
410	* When we spawn Dedicated Sync, we send `$_COOKIES` with the request to help out with any
411	* firewall and/or caching functionality that might prevent us to ping the site directly.
412	*
413	* This will cause Dedicated Sync to reuse the visitor's PHP session and lock it until the
414	* request finishes, which can take anywhere from 1 to 30+ seconds, depending on the server
415	* `max_execution_time` configuration.
416	*
417	* By closing the session we're freeing up the session, so other requests can acquire the
418	* lock and proceed with their own tasks.
419	*/
420	if ( session_status() === PHP_SESSION_ACTIVE ) {
421	session_write_close();
422	}
423
424	// Actually try to send Sync events.
425	$result = $this->do_sync_and_set_delays( $this->sync_queue );
426
427	// Output not used right now. Try to release dedicated sync lock
428	Dedicated_Sender::try_release_lock_spawn_request();
429
430	// If no errors occurred, re-spawn a dedicated Sync request.
431	if ( true === $result ) {
432	Dedicated_Sender::spawn_sync( $this->sync_queue );
433	}
434
435	if ( $do_real_exit ) {
436	exit( 0 );
437	}
438	}
439
440	/**
441	* Trigger sync for a certain sync queue.
442	* Responsible for setting next sync time.
443	* Will not be delayed if the current request is a WP import one.
444	* Will be delayed until the next sync time comes.
445	*
446	* @access public
447	*
448	* @param \Automattic\Jetpack\Sync\Queue $queue Queue object.
449	*
450	* @return boolean\|WP_Error True if this sync sending was successful, error object otherwise.
451	*/
452	public function do_sync_and_set_delays( $queue ) {
453	// Don't sync if importing.
454	if ( defined( 'WP_IMPORTING' ) && WP_IMPORTING ) {
455	return new WP_Error( 'is_importing' );
456	}
457
458	// Don't sync if request is marked as read only.
459	if ( Constants::is_true( 'JETPACK_SYNC_READ_ONLY' ) ) {
460	return new WP_Error( 'jetpack_sync_read_only' );
461	}
462
463	if ( ! Settings::is_sender_enabled( $queue->id ) ) {
464	return new WP_Error( 'sender_disabled_for_queue_' . $queue->id );
465	}
466
467	if ( get_transient( self::TEMP_SYNC_DISABLE_TRANSIENT_NAME ) ) {
468	return new WP_Error( 'sender_temporarily_disabled_while_pulling' );
469	}
470
471	// Return early if we've gotten a retry-after header response.
472	$retry_time = get_option( Actions::RETRY_AFTER_PREFIX . $queue->id );
473	if ( $retry_time ) {
474	// If expired update to false but don't send. Send will occurr in new request to avoid race conditions.
475	if ( microtime( true ) > $retry_time ) {
476	update_option( Actions::RETRY_AFTER_PREFIX . $queue->id, false, false );
477	}
478	return new WP_Error( 'retry_after' );
479	}
480
481	// Don't sync if we are throttled.
482	if ( $this->get_next_sync_time( $queue->id ) > microtime( true ) ) {
483	return new WP_Error( 'sync_throttled' );
484	}
485
486	$start_time = microtime( true );
487
488	Settings::set_is_syncing( true );
489
490	$sync_result = $this->do_sync_for_queue( $queue );
491
492	Settings::set_is_syncing( false );
493
494	$exceeded_sync_wait_threshold = ( microtime( true ) - $start_time ) > (float) $this->get_sync_wait_threshold();
495
496	if ( is_wp_error( $sync_result ) ) {
497	if ( 'unclosed_buffer' === $sync_result->get_error_code() ) {
498	$this->set_next_sync_time( time() + self::QUEUE_LOCKED_SYNC_DELAY, $queue->id );
499	}
500	if ( 'wpcom_error' === $sync_result->get_error_code() ) {
501	$this->set_next_sync_time( time() + self::WPCOM_ERROR_SYNC_DELAY, $queue->id );
502	}
503	} elseif ( $exceeded_sync_wait_threshold && ! Settings::is_doing_cron() ) {
504	// If a send was slow, briefly pause before the next one.
505	// Applies only to Dedicated/Normal Sync to avoid impacting user traffic;
506	// cron jobs are exempt.
507	$this->set_next_sync_time( time() + $this->get_sync_wait_time(), $queue->id );
508	}
509
510	return $sync_result;
511	}
512
513	/**
514	* Retrieve the next sync items to send.
515	*
516	* @access public
517	*
518	* @param (array\|\Automattic\Jetpack\Sync\Queue_Buffer) $buffer_or_items Queue buffer or array of objects.
519	* @param boolean $encode Whether to encode the items.
520	* @return array Sync items to send.
521	*/
522	public function get_items_to_send( $buffer_or_items, $encode = true ) {
523	// Track how long we've been processing so we can avoid request timeouts.
524	$start_time = microtime( true );
525	$upload_size = 0;
526	$items_to_send = array();
527	$items = is_array( $buffer_or_items ) ? $buffer_or_items : $buffer_or_items->get_items();
528	if ( ! is_array( $items ) ) {
529	$items = array();
530	}
531
532	// Set up current screen to avoid errors rendering content.
533	require_once ABSPATH . 'wp-admin/includes/class-wp-screen.php';
534	require_once ABSPATH . 'wp-admin/includes/screen.php';
535	set_current_screen( 'sync' );
536	$skipped_items_ids = array();
537	/**
538	* We estimate the total encoded size as we go by encoding each item individually.
539	* This is expensive, but the only way to really know :/
540	*/
541	foreach ( $items as $key => $item ) {
542	if ( ! is_array( $item ) ) {
543	$skipped_items_ids[] = $key;
544	continue;
545	}
546
547	// Suspending cache addition help prevent overloading in memory cache of large sites.
548	wp_suspend_cache_addition( true );
549	/**
550	* Modify the data within an action before it is serialized and sent to the server
551	* For example, during full sync this expands Post ID's into full Post objects,
552	* so that we don't have to serialize the whole object into the queue.
553	*
554	* @since 1.6.3
555	* @since-jetpack 4.2.0
556	*
557	* @param array The action parameters
558	* @param int The ID of the user who triggered the action
559	*/
560	$item[1] = apply_filters( 'jetpack_sync_before_send_' . $item[0], $item[1], $item[2] );
561	wp_suspend_cache_addition( false );
562	// Serialization usage can lead to empty, null or false action_name. Lets skip as there is no information to send.
563	if ( empty( $item[0] ) \|\| false === $item[1] ) {
564	$skipped_items_ids[] = $key;
565	continue;
566	}
567	$encoded_item = $this->codec->encode( $item );
568	$upload_size += strlen( $encoded_item );
569	if ( $upload_size > $this->upload_max_bytes && array() !== $items_to_send ) {
570	break;
571	}
572	$items_to_send[ $key ] = $encode ? $encoded_item : $item;
573	if ( microtime( true ) - $start_time > $this->max_dequeue_time ) {
574	break;
575	}
576	}
577
578	return array( $items_to_send, $skipped_items_ids, $items, microtime( true ) - $start_time );
579	}
580
581	/**
582	* If supported, flush all response data to the client and finish the request.
583	* This allows for time consuming tasks to be performed without leaving the connection open.
584	*
585	* @access private
586	*/
587	private function fastcgi_finish_request() {
588	if ( function_exists( 'fastcgi_finish_request' ) ) {
589	fastcgi_finish_request();
590	}
591	}
592
593	/**
594	* Perform sync for a certain sync queue.
595	*
596	* @access public
597	*
598	* @param \Automattic\Jetpack\Sync\Queue $queue Queue object.
599	*
600	* @return boolean\|WP_Error True if this sync sending was successful, error object otherwise.
601	*/
602	public function do_sync_for_queue( $queue ) {
603	do_action( 'jetpack_sync_before_send_queue_' . $queue->id );
604	if ( $queue->size() === 0 ) {
605	return new WP_Error( 'empty_queue_' . $queue->id );
606	}
607
608	/**
609	* Now that we're sure we are about to sync, try to ignore user abort
610	* so we can avoid getting into a bad state.
611	*/
612	// https://plugins.trac.wordpress.org/ticket/2041
613	if ( function_exists( 'ignore_user_abort' ) ) {
614	ignore_user_abort( true );
615	}
616
617	/* Don't make the request block till we finish, if possible. */
618	if ( Constants::is_true( 'REST_REQUEST' ) \|\| Constants::is_true( 'XMLRPC_REQUEST' ) ) {
619	$this->fastcgi_finish_request();
620	}
621
622	$checkout_start_time = microtime( true );
623
624	$buffer = $queue->checkout_with_memory_limit( $this->dequeue_max_bytes, $this->upload_max_rows );
625
626	if ( ! $buffer ) {
627	// Buffer has no items.
628	return new WP_Error( 'empty_buffer' );
629	}
630
631	if ( is_wp_error( $buffer ) ) {
632	return $buffer;
633	}
634
635	$checkout_duration = microtime( true ) - $checkout_start_time;
636
637	list( $items_to_send, $skipped_items_ids, $items, $preprocess_duration ) = $this->get_items_to_send( $buffer, true );
638	if ( ! empty( $items_to_send ) ) {
639	/**
640	* Fires when data is ready to send to the server.
641	* Return false or WP_Error to abort the sync (e.g. if there's an error)
642	* The items will be automatically re-sent later
643	*
644	* @since 1.6.3
645	* @since-jetpack 4.2.0
646	*
647	* @param array $data The action buffer
648	* @param string $codec The codec name used to encode the data
649	* @param double $time The current time
650	* @param string $queue The queue used to send ('sync' or 'full_sync')
651	* @param float $checkout_duration The duration of the checkout operation.
652	* @param float $preprocess_duration The duration of the pre-process operation.
653	* @param int $queue_size The size of the sync queue at the time of processing.
654	*/
655	Settings::set_is_sending( true );
656	$processed_item_ids = apply_filters( 'jetpack_sync_send_data', $items_to_send, $this->codec->name(), microtime( true ), $queue->id, $checkout_duration, $preprocess_duration, $queue->size(), $buffer->id );
657	Settings::set_is_sending( false );
658	} else {
659	$processed_item_ids = $skipped_items_ids;
660	$skipped_items_ids = array();
661	}
662
663	if ( 'non-blocking' !== $processed_item_ids ) {
664	if ( ! $processed_item_ids \|\| is_wp_error( $processed_item_ids ) ) {
665	$checked_in_item_ids = $queue->checkin( $buffer );
666	if ( is_wp_error( $checked_in_item_ids ) ) {
667	// phpcs:ignore WordPress.PHP.DevelopmentFunctions.error_log_error_log
668	error_log( 'Error checking in buffer: ' . $checked_in_item_ids->get_error_message() );
669	$queue->force_checkin();
670	}
671	if ( is_wp_error( $processed_item_ids ) ) {
672	return new WP_Error( 'wpcom_error', $processed_item_ids->get_error_code() );
673	}
674
675	// Returning a wpcom_error is a sign to the caller that we should wait a while before syncing again.
676	return new WP_Error( 'wpcom_error', 'jetpack_sync_send_data_false' );
677	} else {
678	// Detect if the last item ID was an error.
679	$had_wp_error = is_wp_error( end( $processed_item_ids ) );
680	$wp_error = $had_wp_error ? array_pop( $processed_item_ids ) : null;
681	// Also checkin any items that were skipped.
682	if ( array() !== $skipped_items_ids ) {
683	$processed_item_ids = array_merge( $processed_item_ids, $skipped_items_ids );
684	}
685	$processed_items = array_intersect_key( $items, array_flip( $processed_item_ids ) );
686	/**
687	* Allows us to keep track of all the actions that have been sent.
688	* Allows us to calculate the progress of specific actions.
689	*
690	* @since 1.6.3
691	* @since-jetpack 4.2.0
692	*
693	* @param array $processed_actions The actions that we send successfully.
694	*/
695	do_action( 'jetpack_sync_processed_actions', $processed_items );
696	$queue->close( $buffer, $processed_item_ids );
697	// Returning a WP_Error is a sign to the caller that we should wait a while before syncing again.
698	if ( $had_wp_error ) {
699	return new WP_Error( 'wpcom_error', $wp_error->get_error_code() );
700	}
701	}
702	}
703
704	return true;
705	}
706
707	/**
708	* Immediately sends a single item without firing or enqueuing it
709	*
710	* @param string $action_name The action.
711	* @param array $data The data associated with the action.
712	* @param string $key The key to use for the action.
713	*
714	* @return array Items processed. TODO: this doesn't make much sense anymore, it should probably be just a bool.
715	*/
716	public function send_action( $action_name, $data = null, $key = null ) {
717	if ( ! Settings::is_sender_enabled( 'full_sync' ) ) {
718	return array();
719	}
720
721	// Compose the data to be sent.
722	$action_to_send = $this->create_action_to_send( $action_name, $data, $key );
723
724	list( $items_to_send, $skipped_items_ids, $items, $preprocess_duration ) = $this->get_items_to_send( $action_to_send, true ); // phpcs:ignore VariableAnalysis.CodeAnalysis.VariableAnalysis.UnusedVariable
725	Settings::set_is_sending( true );
726	$processed_item_ids = apply_filters( 'jetpack_sync_send_data', $items_to_send, $this->get_codec()->name(), microtime( true ), 'immediate-send', 0, $preprocess_duration );
727	Settings::set_is_sending( false );
728
729	/**
730	* Allows us to keep track of all the actions that have been sent.
731	* Allows us to calculate the progress of specific actions.
732	*
733	* @param array $processed_actions The actions that we send successfully.
734	*
735	* @since 1.6.3
736	* @since-jetpack 4.2.0
737	*/
738	do_action( 'jetpack_sync_processed_actions', $action_to_send );
739
740	return $processed_item_ids;
741	}
742
743	/**
744	* Create an synthetic action for direct sending to WPCOM during full sync (for example)
745	*
746	* @access private
747	*
748	* @param string $action_name The action.
749	* @param array $data The data associated with the action.
750	* @param string $key The key to use for the action.
751	* @return array An array of synthetic sync actions keyed by current microtime(true)
752	*/
753	private function create_action_to_send( $action_name, $data, $key = null ) {
754	return array(
755	$key ?? (string) microtime( true ) => array(
756	$action_name,
757	$data,
758	get_current_user_id(),
759	microtime( true ),
760	Settings::is_importing(),
761	),
762	);
763	}
764
765	/**
766	* Returns any object that is able to be synced.
767	*
768	* @access public
769	*
770	* @param array $args the synchronized object parameters.
771	* @return string Encoded sync object.
772	*/
773	public function sync_object( $args ) {
774	// For example: posts, post, 5.
775	list( $module_name, $object_type, $id ) = $args;
776
777	$sync_module = Modules::get_module( $module_name );
778	$codec = $this->get_codec();
779
780	return $codec->encode( $sync_module->get_object_by_id( $object_type, $id ) );
781	}
782
783	/**
784	* Register additional sync XML-RPC methods available to Jetpack for authenticated users.
785	*
786	* @access public
787	* @since 1.6.3
788	* @since-jetpack 7.8.0
789	*
790	* @param array $jetpack_methods XML-RPC methods available to the Jetpack Server.
791	* @return array Filtered XML-RPC methods.
792	*/
793	public function register_jetpack_xmlrpc_methods( $jetpack_methods ) {
794	$jetpack_methods['jetpack.syncObject'] = array( $this, 'sync_object' );
795	return $jetpack_methods;
796	}
797
798	/**
799	* Get the incremental sync queue object.
800	*
801	* @access public
802	*
803	* @return \Automattic\Jetpack\Sync\Queue Queue object.
804	*/
805	public function get_sync_queue() {
806	return $this->sync_queue;
807	}
808
809	/**
810	* Get the full sync queue object.
811	*
812	* @access public
813	*
814	* @return \Automattic\Jetpack\Sync\Queue Queue object.
815	*/
816	public function get_full_sync_queue() {
817	return $this->full_sync_queue;
818	}
819
820	/**
821	* Get the codec object.
822	*
823	* @access public
824	*
825	* @return \Automattic\Jetpack\Sync\Codec_Interface Codec object.
826	*/
827	public function get_codec() {
828	return $this->codec;
829	}
830
831	/**
832	* Determine the codec object.
833	* Use gzip deflate if supported.
834	*
835	* @access public
836	*/
837	public function set_codec() {
838	if ( function_exists( 'gzinflate' ) ) {
839	$this->codec = new JSON_Deflate_Array_Codec();
840	} else {
841	$this->codec = new Simple_Codec();
842	}
843	}
844
845	/**
846	* Compute and send all the checksums.
847	*
848	* @access public
849	*/
850	public function send_checksum() {
851	$store = new Replicastore();
852	do_action( 'jetpack_sync_checksum', $store->checksum_all() );
853	}
854
855	/**
856	* Reset the incremental sync queue.
857	*
858	* @access public
859	*/
860	public function reset_sync_queue() {
861	$this->sync_queue->reset();
862	}
863
864	/**
865	* Reset the full sync queue.
866	*
867	* @access public
868	*/
869	public function reset_full_sync_queue() {
870	$this->full_sync_queue->reset();
871	}
872
873	/**
874	* Set the maximum bytes to checkout without exceeding the memory limit.
875	*
876	* @access public
877	*
878	* @param int $size Maximum bytes to checkout.
879	*/
880	public function set_dequeue_max_bytes( $size ) {
881	$this->dequeue_max_bytes = $size;
882	}
883
884	/**
885	* Set the maximum bytes in a single encoded item.
886	*
887	* @access public
888	*
889	* @param int $max_bytes Maximum bytes in a single encoded item.
890	*/
891	public function set_upload_max_bytes( $max_bytes ) {
892	$this->upload_max_bytes = $max_bytes;
893	}
894
895	/**
896	* Set the maximum number of sync items in a single action.
897	*
898	* @access public
899	*
900	* @param int $max_rows Maximum number of sync items.
901	*/
902	public function set_upload_max_rows( $max_rows ) {
903	$this->upload_max_rows = $max_rows;
904	}
905
906	/**
907	* Set the sync wait time (in seconds).
908	*
909	* @access public
910	*
911	* @param int $seconds Sync wait time.
912	*/
913	public function set_sync_wait_time( $seconds ) {
914	$this->sync_wait_time = $seconds;
915	}
916
917	/**
918	* Get current sync wait time (in seconds).
919	*
920	* @access public
921	*
922	* @return int Sync wait time.
923	*/
924	public function get_sync_wait_time() {
925	return $this->sync_wait_time;
926	}
927
928	/**
929	* Set the enqueue wait time (in seconds).
930	*
931	* @access public
932	*
933	* @param int $seconds Enqueue wait time.
934	*/
935	public function set_enqueue_wait_time( $seconds ) {
936	$this->enqueue_wait_time = $seconds;
937	}
938
939	/**
940	* Get current enqueue wait time (in seconds).
941	*
942	* @access public
943	*
944	* @return int Enqueue wait time.
945	*/
946	public function get_enqueue_wait_time() {
947	return $this->enqueue_wait_time;
948	}
949
950	/**
951	* Set the sync wait threshold (in seconds).
952	*
953	* @access public
954	*
955	* @param int $seconds Sync wait threshold.
956	*/
957	public function set_sync_wait_threshold( $seconds ) {
958	$this->sync_wait_threshold = $seconds;
959	}
960
961	/**
962	* Get current sync wait threshold (in seconds).
963	*
964	* @access public
965	*
966	* @return int Sync wait threshold.
967	*/
968	public function get_sync_wait_threshold() {
969	return $this->sync_wait_threshold;
970	}
971
972	/**
973	* Set the maximum time for perfirming a checkout of items from the queue (in seconds).
974	*
975	* @access public
976	*
977	* @param int $seconds Maximum dequeue time.
978	*/
979	public function set_max_dequeue_time( $seconds ) {
980	$this->max_dequeue_time = $seconds;
981	}
982
983	/**
984	* Initialize the sync queues, codec and set the default settings.
985	*
986	* @access public
987	*/
988	public function set_defaults() {
989	$this->sync_queue = new Queue( 'sync' );
990	$this->full_sync_queue = new Queue( 'full_sync' );
991	$this->set_codec();
992
993	// Saved settings.
994	Settings::set_importing( null );
995	$settings = Settings::get_settings();
996	$this->set_dequeue_max_bytes( $settings['dequeue_max_bytes'] );
997	$this->set_upload_max_bytes( $settings['upload_max_bytes'] );
998	$this->set_upload_max_rows( $settings['upload_max_rows'] );
999	$this->set_sync_wait_time( $settings['sync_wait_time'] );
1000	$this->set_enqueue_wait_time( $settings['enqueue_wait_time'] );
1001	$this->set_sync_wait_threshold( $settings['sync_wait_threshold'] );
1002	$this->set_max_dequeue_time( Defaults::get_max_sync_execution_time() );
1003	}
1004
1005	/**
1006	* Reset sync queues, modules and settings.
1007	*
1008	* @access public
1009	*/
1010	public function reset_data() {
1011	$this->reset_sync_queue();
1012	$this->reset_full_sync_queue();
1013
1014	foreach ( Modules::get_modules() as $module ) {
1015	$module->reset_data();
1016	}
1017	// Reset Sync locks without unlocking queues since we already reset those.
1018	Actions::reset_sync_locks( false );
1019
1020	Settings::reset_data();
1021	}
1022
1023	/**
1024	* Perform cleanup at the event of plugin uninstallation.
1025	*
1026	* @access public
1027	*/
1028	public function uninstall() {
1029	// Lets delete all the other fun stuff like transient and option and the sync queue.
1030	$this->reset_data();
1031
1032	// Delete the full sync status.
1033	delete_option( 'jetpack_full_sync_status' );
1034
1035	// Clear the sync cron.
1036	wp_clear_scheduled_hook( 'jetpack_sync_cron' );
1037	wp_clear_scheduled_hook( 'jetpack_sync_full_cron' );
1038	}
1039	}
1040