[C2S] Add Server-Sent Events (SSE) for real-time collection streaming

activitypub.php

@@ -74,6 +74,7 @@ function rest_init() {
 	// Load OAuth REST endpoints.
 	( new Rest\OAuth_Controller() )->register_routes();
 	( new Rest\Proxy_Controller() )->register_routes();
+	( new Rest\Event_Stream_Controller() )->register_routes();
 }
 \add_action( 'rest_api_init', __NAMESPACE__ . '\rest_init' );
 
@@ -87,21 +88,22 @@ function plugin_init() {
 	\add_action( 'init', array( __NAMESPACE__ . '\Comment', 'init' ) );
 	\add_action( 'init', array( __NAMESPACE__ . '\Dispatcher', 'init' ) );
 	\add_action( 'init', array( __NAMESPACE__ . '\Embed', 'init' ) );
+	\add_action( 'init', array( __NAMESPACE__ . '\Event_Stream', 'init' ) );
 	\add_action( 'init', array( __NAMESPACE__ . '\Handler', 'init' ) );
 	\add_action( 'init', array( __NAMESPACE__ . '\Hashtag', 'init' ) );
 	\add_action( 'init', array( __NAMESPACE__ . '\Link', 'init' ) );
 	\add_action( 'init', array( __NAMESPACE__ . '\Mailer', 'init' ) );
 	\add_action( 'init', array( __NAMESPACE__ . '\Mention', 'init' ) );
 	\add_action( 'init', array( __NAMESPACE__ . '\Migration', 'init' ), 1 );
 	\add_action( 'init', array( __NAMESPACE__ . '\Move', 'init' ) );
+	\add_action( 'init', array( __NAMESPACE__ . '\OAuth\Server', 'init' ) );
 	\add_action( 'init', array( __NAMESPACE__ . '\Options', 'init' ) );
 	\add_action( 'init', array( __NAMESPACE__ . '\Post_Types', 'init' ) );
 	\add_action( 'init', array( __NAMESPACE__ . '\Router', 'init' ) );
 	// Priority 0 ensures Scheduler hooks are registered before Migration (priority 1) runs.
 	\add_action( 'init', array( __NAMESPACE__ . '\Scheduler', 'init' ), 0 );
 	\add_action( 'init', array( __NAMESPACE__ . '\Search', 'init' ) );
 	\add_action( 'init', array( __NAMESPACE__ . '\Signature', 'init' ) );
-	\add_action( 'init', array( __NAMESPACE__ . '\OAuth\Server', 'init' ) );
 
 	if ( site_supports_blocks() ) {
 		\add_action( 'init', array( __NAMESPACE__ . '\Blocks', 'init' ) );

includes/class-event-stream.php

@@ -0,0 +1,61 @@
+<?php
+/**
+ * Event Stream signal writer.
+ *
+ * Sets lightweight transient signals when new items arrive in
+ * outbox or inbox collections, so the SSE polling loop can avoid
+ * unnecessary database queries.
+ *
+ * @package Activitypub
+ * @see https://swicg.github.io/activitypub-api/sse
+ * @since unreleased
+ */
+
+namespace Activitypub;
+
+/**
+ * Event_Stream class.
+ *
+ * Hooks into outbox and inbox actions to set transient signals
+ * that the SSE controller checks during its polling loop.
+ *
+ * @since unreleased
+ */
+class Event_Stream {
+	/**
+	 * Initialize the event stream signals.
+	 */
+	public static function init() {
+		\add_action( 'post_activitypub_add_to_outbox', array( self::class, 'signal_outbox' ), 10, 3 );
+		\add_action( 'activitypub_handled_inbox', array( self::class, 'signal_inbox' ), 10, 2 );
+	}
+
+	/**
+	 * Set a transient signal when a new item is added to the outbox.
+	 *
+	 * @param int                            $outbox_activity_id The outbox post ID.
+	 * @param \Activitypub\Activity\Activity $activity           The activity object.
+	 * @param int                            $user_id            The user ID.
+	 */
+	public static function signal_outbox( $outbox_activity_id, $activity, $user_id ) { // phpcs:ignore VariableAnalysis.CodeAnalysis.VariableAnalysis.UnusedVariable
+		$signal_key = sprintf( 'activitypub_sse_signal_%s_outbox', $user_id );
+		\set_transient( $signal_key, time(), HOUR_IN_SECONDS );
+	}
+
+	/**
+	 * Set transient signals when a new item is added to the inbox.
+	 *
+	 * @param array $data     The activity data array.
+	 * @param array $user_ids The user IDs that received the activity.
+	 */
+	public static function signal_inbox( $data, $user_ids ) { // phpcs:ignore VariableAnalysis.CodeAnalysis.VariableAnalysis.UnusedVariable
+		if ( ! \is_array( $user_ids ) ) {
+			$user_ids = array( $user_ids );
+		}
+
+		foreach ( $user_ids as $user_id ) {
+			$signal_key = sprintf( 'activitypub_sse_signal_%s_inbox', $user_id );
+			\set_transient( $signal_key, time(), HOUR_IN_SECONDS );
+		}
+	}
+}

includes/model/class-blog.php

@@ -401,6 +401,7 @@ public function get_endpoints() {
 			'oauthAuthorizationEndpoint' => get_rest_url_by_path( 'oauth/authorize' ),
 			'oauthTokenEndpoint'         => get_rest_url_by_path( 'oauth/token' ),
 			'proxyUrl'                   => get_rest_url_by_path( 'proxy' ),
+			'proxyEventStream'           => get_rest_url_by_path( 'proxy-event-stream' ),
 		);
 	}
 

includes/model/class-user.php

@@ -322,6 +322,7 @@ public function get_endpoints() {
 			'oauthAuthorizationEndpoint' => get_rest_url_by_path( 'oauth/authorize' ),
 			'oauthTokenEndpoint'         => get_rest_url_by_path( 'oauth/token' ),
 			'proxyUrl'                   => get_rest_url_by_path( 'proxy' ),
+			'proxyEventStream'           => get_rest_url_by_path( 'proxy-event-stream' ),
 		);
 	}
 

includes/oauth/class-server.php

@@ -121,16 +121,24 @@ public static function has_scope( $scope ) {
 	public static function get_bearer_token() {
 		$auth_header = self::get_authorization_header();
 
-		if ( ! $auth_header ) {
-			return null;
+		if ( $auth_header && 0 === strpos( $auth_header, 'Bearer ' ) ) {
+			return substr( $auth_header, 7 );
 		}
 
-		// Check for Bearer token.
-		if ( 0 !== strpos( $auth_header, 'Bearer ' ) ) {
-			return null;
+		/*
+		 * Fall back to `access_token` query parameter for EventSource clients.
+		 * The browser EventSource API cannot send custom headers, so the SSE
+		 * spec requires accepting the token as a query parameter.
+		 *
+		 * @see https://swicg.github.io/activitypub-api/sse
+		 */
+		// phpcs:disable WordPress.Security.NonceVerification.Recommended, WordPress.Security.ValidatedSanitizedInput.InputNotSanitized -- Opaque auth token, must not be altered.
+		if ( ! empty( $_GET['access_token'] ) ) {
+			return \wp_unslash( $_GET['access_token'] );
 		}
+		// phpcs:enable WordPress.Security.NonceVerification.Recommended, WordPress.Security.ValidatedSanitizedInput.InputNotSanitized
 
-		return substr( $auth_header, 7 );
+		return null;
 	}
 
 	/**

includes/rest/class-actors-inbox-controller.php

@@ -167,6 +167,7 @@ public function get_items( $request ) {
 			'actor'        => $user->get_id(),
 			'type'         => 'OrderedCollection',
 			'totalItems'   => (int) $inbox_query->found_posts,
+			'eventStream'  => Event_Stream_Controller::get_stream_url( $user_id, 'inbox' ),
 			'orderedItems' => array(),
 		);