activitypub / includes / class-scheduler.php

class-scheduler.php in ActivityPub 5.7.0, at includes/class-scheduler.php

445 lines 12.0 KB

Raw Download Zip

1	<?php
2	/**
3	* Scheduler class file.
4	*
5	* @package Activitypub
6	*/
7
8	namespace Activitypub;
9
10	use Activitypub\Activity\Activity;
11	use Activitypub\Activity\Base_Object;
12	use Activitypub\Scheduler\Post;
13	use Activitypub\Scheduler\Actor;
14	use Activitypub\Scheduler\Comment;
15	use Activitypub\Collection\Actors;
16	use Activitypub\Collection\Outbox;
17	use Activitypub\Collection\Followers;
18	use Activitypub\Transformer\Factory;
19
20	/**
21	* Scheduler class.
22	*
23	* @author Matthias Pfefferle
24	*/
25	class Scheduler {
26
27	/**
28	* Allowed batch callbacks.
29	*
30	* @var array
31	*/
32	private static $batch_callbacks = array();
33
34	/**
35	* Initialize the class, registering WordPress hooks.
36	*/
37	public static function init() {
38	self::register_schedulers();
39
40	self::$batch_callbacks = array(
41	Dispatcher::$callback,
42	array( Dispatcher::class, 'retry_send_to_followers' ),
43	);
44
45	// Follower Cleanups.
46	\add_action( 'activitypub_update_followers', array( self::class, 'update_followers' ) );
47	\add_action( 'activitypub_cleanup_followers', array( self::class, 'cleanup_followers' ) );
48
49	// Event callbacks.
50	\add_action( 'activitypub_async_batch', array( self::class, 'async_batch' ), 10, 99 );
51	\add_action( 'activitypub_reprocess_outbox', array( self::class, 'reprocess_outbox' ) );
52	\add_action( 'activitypub_outbox_purge', array( self::class, 'purge_outbox' ) );
53
54	\add_action( 'post_activitypub_add_to_outbox', array( self::class, 'schedule_outbox_activity_for_federation' ) );
55	\add_action( 'post_activitypub_add_to_outbox', array( self::class, 'schedule_announce_activity' ), 10, 4 );
56
57	\add_action( 'update_option_activitypub_outbox_purge_days', array( self::class, 'handle_outbox_purge_days_update' ), 10, 2 );
58	}
59
60	/**
61	* Register handlers.
62	*/
63	public static function register_schedulers() {
64	Post::init();
65	Actor::init();
66	Comment::init();
67
68	/**
69	* Register additional schedulers.
70	*
71	* @since 5.0.0
72	*/
73	do_action( 'activitypub_register_schedulers' );
74	}
75
76	/**
77	* Schedule all ActivityPub schedules.
78	*/
79	public static function register_schedules() {
80	if ( ! \wp_next_scheduled( 'activitypub_update_followers' ) ) {
81	\wp_schedule_event( time(), 'hourly', 'activitypub_update_followers' );
82	}
83
84	if ( ! \wp_next_scheduled( 'activitypub_cleanup_followers' ) ) {
85	\wp_schedule_event( time(), 'daily', 'activitypub_cleanup_followers' );
86	}
87
88	if ( ! \wp_next_scheduled( 'activitypub_reprocess_outbox' ) ) {
89	\wp_schedule_event( time(), 'hourly', 'activitypub_reprocess_outbox' );
90	}
91
92	if ( ! wp_next_scheduled( 'activitypub_outbox_purge' ) ) {
93	wp_schedule_event( time(), 'daily', 'activitypub_outbox_purge' );
94	}
95	}
96
97	/**
98	* Un-schedule all ActivityPub schedules.
99	*
100	* @return void
101	*/
102	public static function deregister_schedules() {
103	wp_unschedule_hook( 'activitypub_update_followers' );
104	wp_unschedule_hook( 'activitypub_cleanup_followers' );
105	wp_unschedule_hook( 'activitypub_reprocess_outbox' );
106	wp_unschedule_hook( 'activitypub_outbox_purge' );
107	}
108
109	/**
110	* Update followers.
111	*/
112	public static function update_followers() {
113	$number = 5;
114
115	if ( defined( 'DISABLE_WP_CRON' ) && DISABLE_WP_CRON ) {
116	$number = 50;
117	}
118
119	/**
120	* Filter the number of followers to update.
121	*
122	* @param int $number The number of followers to update.
123	*/
124	$number = apply_filters( 'activitypub_update_followers_number', $number );
125	$followers = Followers::get_outdated_followers( $number );
126
127	foreach ( $followers as $follower ) {
128	$meta = get_remote_metadata_by_actor( $follower->get_id(), false );
129
130	if ( empty( $meta ) \|\| ! is_array( $meta ) \|\| is_wp_error( $meta ) ) {
131	Followers::add_error( $follower->get__id(), $meta );
132	} else {
133	$follower->from_array( $meta );
134	$follower->update();
135	}
136	}
137	}
138
139	/**
140	* Cleanup followers.
141	*/
142	public static function cleanup_followers() {
143	$number = 5;
144
145	if ( defined( 'DISABLE_WP_CRON' ) && DISABLE_WP_CRON ) {
146	$number = 50;
147	}
148
149	/**
150	* Filter the number of followers to clean up.
151	*
152	* @param int $number The number of followers to clean up.
153	*/
154	$number = apply_filters( 'activitypub_update_followers_number', $number );
155	$followers = Followers::get_faulty_followers( $number );
156
157	foreach ( $followers as $follower ) {
158	$meta = get_remote_metadata_by_actor( $follower->get_url(), false );
159
160	if ( is_tombstone( $meta ) ) {
161	$follower->delete();
162	} elseif ( empty( $meta ) \|\| ! is_array( $meta ) \|\| is_wp_error( $meta ) ) {
163	if ( $follower->count_errors() >= 5 ) {
164	$follower->delete();
165	\wp_schedule_single_event(
166	\time(),
167	'activitypub_delete_actor_interactions',
168	array( $follower->get_id() )
169	);
170	} else {
171	Followers::add_error( $follower->get__id(), $meta );
172	}
173	} else {
174	$follower->reset_errors();
175	}
176	}
177	}
178
179	/**
180	* Schedule the outbox item for federation.
181	*
182	* @param int $id The ID of the outbox item.
183	* @param int $offset The offset to add to the scheduled time.
184	*/
185	public static function schedule_outbox_activity_for_federation( $id, $offset = 0 ) {
186	$hook = 'activitypub_process_outbox';
187	$args = array( $id );
188
189	if ( false === wp_next_scheduled( $hook, $args ) ) {
190	\wp_schedule_single_event(
191	\time() + $offset,
192	$hook,
193	$args
194	);
195	}
196	}
197
198	/**
199	* Reprocess the outbox.
200	*/
201	public static function reprocess_outbox() {
202	// Bail if there is a pending batch.
203	if ( self::next_scheduled_hook( 'activitypub_async_batch' ) ) {
204	return;
205	}
206
207	// Bail if there is a batch in progress.
208	$key = \md5( \serialize( Dispatcher::$callback ) ); // phpcs:ignore WordPress.PHP.DiscouragedPHPFunctions.serialize_serialize
209	if ( self::is_locked( $key ) ) {
210	return;
211	}
212
213	$ids = \get_posts(
214	array(
215	'post_type' => Outbox::POST_TYPE,
216	'post_status' => 'pending',
217	'posts_per_page' => 10,
218	'fields' => 'ids',
219	)
220	);
221
222	foreach ( $ids as $id ) {
223	self::schedule_outbox_activity_for_federation( $id );
224	}
225	}
226
227	/**
228	* Purge outbox items based on a schedule.
229	*/
230	public static function purge_outbox() {
231	$total_posts = (int) wp_count_posts( Outbox::POST_TYPE )->publish;
232	if ( $total_posts <= 20 ) {
233	return;
234	}
235
236	$days = (int) get_option( 'activitypub_outbox_purge_days', 180 );
237	$timezone = new \DateTimeZone( 'UTC' );
238	$date = new \DateTime( 'now', $timezone );
239
240	$date->sub( \DateInterval::createFromDateString( "$days days" ) );
241
242	$post_ids = get_posts(
243	array(
244	'post_type' => Outbox::POST_TYPE,
245	'post_status' => 'any',
246	'fields' => 'ids',
247	'numberposts' => -1,
248	'date_query' => array(
249	array(
250	'before' => $date->format( 'Y-m-d' ),
251	),
252	),
253	)
254	);
255
256	foreach ( $post_ids as $post_id ) {
257	\wp_delete_post( $post_id, true );
258	}
259	}
260
261	/**
262	* Update schedules when outbox purge days settings change.
263	*
264	* @param int $old_value The old value.
265	* @param int $value The new value.
266	*/
267	public static function handle_outbox_purge_days_update( $old_value, $value ) {
268	if ( 0 === (int) $value ) {
269	wp_clear_scheduled_hook( 'activitypub_outbox_purge' );
270	} elseif ( ! wp_next_scheduled( 'activitypub_outbox_purge' ) ) {
271	wp_schedule_event( time(), 'daily', 'activitypub_outbox_purge' );
272	}
273	}
274
275	/**
276	* Asynchronously runs batch processing routines.
277	*
278	* The batching part is optional and only comes into play if the callback returns anything.
279	* Beyond that it's a helper to run a callback asynchronously with locking to prevent simultaneous processing.
280	*
281	* @param callable $callback Callable processing routine.
282	* @params mixed ...$args Optional. Parameters that get passed to the callback.
283	*/
284	public static function async_batch( $callback ) {
285	if ( ! in_array( $callback, self::$batch_callbacks, true ) \|\| ! \is_callable( $callback ) ) {
286	_doing_it_wrong( __METHOD__, 'The first argument must be a valid callback.', '5.2.0' );
287	return;
288	}
289
290	$args = \func_get_args(); // phpcs:ignore PHPCompatibility.FunctionUse.ArgumentFunctionsReportCurrentValue
291	$key = \md5( \serialize( $callback ) ); // phpcs:ignore WordPress.PHP.DiscouragedPHPFunctions.serialize_serialize
292
293	// Bail if the existing lock is still valid.
294	if ( self::is_locked( $key ) ) {
295	\wp_schedule_single_event( time() + MINUTE_IN_SECONDS, 'activitypub_async_batch', $args );
296	return;
297	}
298
299	self::lock( $key );
300
301	$callback = array_shift( $args ); // Remove $callback from arguments.
302	$next = \call_user_func_array( $callback, $args );
303
304	self::unlock( $key );
305
306	if ( ! empty( $next ) ) {
307	// Schedule the next run, adding the result to the arguments.
308	\wp_schedule_single_event(
309	\time() + 30,
310	'activitypub_async_batch',
311	\array_merge( array( $callback ), \array_values( $next ) )
312	);
313	}
314	}
315
316
317	/**
318	* Locks the async batch process for individual callbacks to prevent simultaneous processing.
319	*
320	* @param string $key Serialized callback name.
321	* @return bool\|int True if the lock was successful, timestamp of existing lock otherwise.
322	*/
323	public static function lock( $key ) {
324	global $wpdb;
325
326	// Try to lock.
327	$lock_result = (bool) $wpdb->query( $wpdb->prepare( "INSERT IGNORE INTO `$wpdb->options` ( `option_name`, `option_value`, `autoload` ) VALUES (%s, %s, 'no') /* LOCK */", 'activitypub_async_batch_' . $key, \time() ) ); // phpcs:ignore WordPress.DB
328
329	if ( ! $lock_result ) {
330	$lock_result = \get_option( 'activitypub_async_batch_' . $key );
331	}
332
333	return $lock_result;
334	}
335
336	/**
337	* Unlocks processing for the async batch callback.
338	*
339	* @param string $key Serialized callback name.
340	*/
341	public static function unlock( $key ) {
342	\delete_option( 'activitypub_async_batch_' . $key );
343	}
344
345	/**
346	* Whether the async batch callback is locked.
347	*
348	* @param string $key Serialized callback name.
349	* @return boolean
350	*/
351	public static function is_locked( $key ) {
352	$lock = \get_option( 'activitypub_async_batch_' . $key );
353
354	if ( ! $lock ) {
355	return false;
356	}
357
358	$lock = (int) $lock;
359
360	if ( $lock < \time() - 1800 ) {
361	self::unlock( $key );
362	return false;
363	}
364
365	return true;
366	}
367
368	/**
369	* Get the next scheduled hook.
370	*
371	* @param string $hook The hook name.
372	* @return int\|bool The timestamp of the next scheduled hook, or false if none found.
373	*/
374	private static function next_scheduled_hook( $hook ) {
375	$crons = _get_cron_array();
376	if ( empty( $crons ) ) {
377	return false;
378	}
379
380	// Get next event.
381	$next = false;
382	foreach ( $crons as $timestamp => $cron ) {
383	if ( isset( $cron[ $hook ] ) ) {
384	$next = $timestamp;
385	break;
386	}
387	}
388
389	return $next;
390	}
391
392	/**
393	* Send announces.
394	*
395	* @param int $outbox_activity_id The outbox activity ID.
396	* @param \Activitypub\Activity\Activity $activity The activity object.
397	* @param int $actor_id The actor ID.
398	* @param int $content_visibility The content visibility.
399	*/
400	public static function schedule_announce_activity( $outbox_activity_id, $activity, $actor_id, $content_visibility ) {
401	// Only if we're in both Blog and User modes.
402	if ( ACTIVITYPUB_ACTOR_AND_BLOG_MODE !== \get_option( 'activitypub_actor_mode', ACTIVITYPUB_ACTOR_MODE ) ) {
403	return;
404	}
405
406	// Only if this isn't the Blog Actor.
407	if ( Actors::BLOG_USER_ID === $actor_id ) {
408	return;
409	}
410
411	// Only if the content is public or quiet public.
412	if ( ACTIVITYPUB_CONTENT_VISIBILITY_PUBLIC !== $content_visibility ) {
413	return;
414	}
415
416	// Only if the activity is a Create.
417	if ( 'Create' !== $activity->get_type() ) {
418	return;
419	}
420
421	if ( ! is_object( $activity->get_object() ) ) {
422	return;
423	}
424
425	// Check if the object is an article, image, audio, video, event, or document and ignore profile updates and other activities.
426	if ( ! in_array( $activity->get_object()->get_type(), Base_Object::TYPES, true ) ) {
427	return;
428	}
429
430	$announce = new Activity();
431	$announce->set_type( 'Announce' );
432	$announce->set_actor( Actors::get_by_id( Actors::BLOG_USER_ID )->get_id() );
433	$announce->set_object( $activity );
434
435	$outbox_activity_id = Outbox::add( $announce, Actors::BLOG_USER_ID );
436
437	if ( ! $outbox_activity_id ) {
438	return;
439	}
440
441	// Schedule the outbox item for federation.
442	self::schedule_outbox_activity_for_federation( $outbox_activity_id, 120 );
443	}
444	}
445