<?php
/**
 * Event Stream Trait file.
 *
 * Provides Server-Sent Events (SSE) functionality for real-time
 * ActivityPub collection updates and proxy streaming.
 *
 * @package Activitypub
 * @see https://swicg.github.io/activitypub-api/sse
 * @since 8.1.0
 */

namespace Activitypub\Rest;

use Activitypub\Collection\Actors;
use Activitypub\Collection\Inbox;
use Activitypub\Collection\Outbox;
use Activitypub\OAuth\Scope;
use Activitypub\OAuth\Server as OAuth_Server;

use function Activitypub\resolve_public_host;


/**
 * Event Stream Trait.
 *
 * Provides SSE streaming capabilities for collection controllers
 * (Outbox, Inbox) and the Proxy controller.
 *
 * @since 8.1.0
 */
trait Event_Stream {

	/**
	 * Map of ActivityPub activity types to SSE event types.
	 *
	 * @see https://swicg.github.io/activitypub-api/sse
	 *
	 * @return array The event type map.
	 */
	protected static function get_event_type_map() {
		return array(
			'Create'   => 'Add',
			'Announce' => 'Add',
			'Like'     => 'Add',
			'Update'   => 'Update',
			'Delete'   => 'Delete',
			'Undo'     => 'Remove',
		);
	}

	/**
	 * Check permissions for the stream endpoint.
	 *
	 * Requires OAuth authentication with the push scope.
	 * Falls back to `access_token` query parameter for EventSource clients,
	 * since the browser EventSource API cannot send custom headers.
	 *
	 * @see https://swicg.github.io/activitypub-api/sse
	 *
	 * @param \WP_REST_Request $request Full details about the request.
	 *
	 * @return bool|\WP_Error True if authorized, WP_Error otherwise.
	 */
	public function get_stream_permissions_check( $request ) {
		// If not already OAuth-authenticated, try the access_token query parameter.
		if ( ! OAuth_Server::is_oauth_request() ) {
			$this->authenticate_from_query_param();
		}

		$oauth_result = OAuth_Server::check_oauth_permission( $request, Scope::PUSH );

		if ( true !== $oauth_result ) {
			return $oauth_result;
		}

		$user_id = $request->get_param( 'user_id' );

		if ( null === $user_id ) {
			return true;
		}

		return $this->verify_owner( $request );
	}

	/**
	 * Authenticate from the access_token query parameter.
	 *
	 * The browser EventSource API cannot send custom headers, so SSE
	 * clients pass the OAuth token as a query parameter. This method
	 * injects it as an Authorization header and re-runs OAuth
	 * authentication so the server recognizes the request.
	 *
	 * @since 8.1.0
	 *
	 * @see https://swicg.github.io/activitypub-api/sse
	 */
	private function authenticate_from_query_param() {
		// phpcs:disable WordPress.Security.NonceVerification.Recommended, WordPress.Security.ValidatedSanitizedInput.InputNotSanitized -- Opaque auth token, must not be altered.
		if ( empty( $_GET['access_token'] ) || ! \is_string( $_GET['access_token'] ) ) {
			return;
		}

		$token_string = \wp_unslash( $_GET['access_token'] );
		// phpcs:enable WordPress.Security.NonceVerification.Recommended, WordPress.Security.ValidatedSanitizedInput.InputNotSanitized

		// Reject tokens that are too long or contain unexpected characters.
		if ( \strlen( $token_string ) > 512 || \preg_match( '/[^A-Za-z0-9._~+\/-]/', $token_string ) ) {
			return;
		}

		// Inject as Authorization header so the OAuth server can find it.
		$_SERVER['HTTP_AUTHORIZATION'] = 'Bearer ' . $token_string;

		// Re-run OAuth authentication.
		OAuth_Server::authenticate_oauth( null );
	}

	/**
	 * Stream SSE events for a collection.
	 *
	 * Sends raw SSE output and calls exit.
	 *
	 * @param int    $user_id    The actor ID.
	 * @param string $collection The collection type ('outbox' or 'inbox').
	 */
	protected function stream_collection( $user_id, $collection ) {
		// Allow PHP to detect client disconnects instead of auto-terminating.
		ignore_user_abort( true );

		// Extend PHP execution time for long-lived SSE connections.
		set_time_limit( 0 );

		$this->send_sse_headers();

		// Honor Last-Event-ID for reconnecting clients (per SSE spec).
		$last_event_id = isset( $_SERVER['HTTP_LAST_EVENT_ID'] )
			? \absint( \wp_unslash( $_SERVER['HTTP_LAST_EVENT_ID'] ) )
			: 0;

		// Use Last-Event-ID if provided, otherwise start from the latest item.
		$since_id = $last_event_id ? $last_event_id : $this->get_latest_item_id( $user_id, $collection );
		$start    = time();

		$this->send_sse_comment( 'connected' );

		while ( ( time() - $start ) < 300 ) {
			if ( \connection_aborted() ) {
				break;
			}

			// Check for signal transient before querying the DB.
			$signal_key = sprintf( 'activitypub_sse_signal_%s_%s', $user_id, $collection );
			$signal     = \get_transient( $signal_key );

			if ( $signal ) {
				\delete_transient( $signal_key );

				$new_items = $this->get_new_items( $user_id, $collection, $since_id );

				foreach ( $new_items as $item ) {
					$this->send_sse_event( $item, $collection );

					if ( $item->ID > $since_id ) {
						$since_id = $item->ID;
					}
				}

				// Re-set signal if we hit the limit, so remaining items are fetched next iteration.
				if ( count( $new_items ) >= 20 ) {
					\set_transient( $signal_key, time(), 5 * MINUTE_IN_SECONDS );
				}
			}

			$this->send_sse_comment( 'keepalive ' . \gmdate( 'c' ) );
			$this->flush_output();

			// phpcs:ignore WordPress.WP.AlternativeFunctions.sleep_sleep -- SSE long-polling requires blocking sleep.
			sleep( 5 );
		}

		$this->send_sse_comment( 'timeout' );
		$this->flush_output();

		exit;
	}

	/**
	 * Open a streaming connection to a remote SSE endpoint and relay events.
	 *
	 * Uses PHP streams directly because the WordPress HTTP API
	 * does not support streaming responses.
	 *
	 * @param string $stream_url The remote eventStream URL.
	 */
	protected function relay_remote_stream( $stream_url ) {
		ignore_user_abort( true );

		// Extend PHP execution time for long-lived SSE connections.
		set_time_limit( 0 );

		$parsed = \wp_parse_url( $stream_url );
		$host   = $parsed['host'];
		$port   = isset( $parsed['port'] ) ? $parsed['port'] : 443;
		$path   = isset( $parsed['path'] ) ? $parsed['path'] : '/';

		if ( isset( $parsed['query'] ) ) {
			$path .= '?' . $parsed['query'];
		}

		/*
		 * Resolve the host once and verify every returned address is public.
		 * stream_socket_client would otherwise do its own lookup, which opens
		 * a DNS-rebinding window after validate_url() already passed.
		 */
		$ip = resolve_public_host( $host );
		if ( false === $ip ) {
			\status_header( 502 );
			\header( 'Content-Type: application/json' );
			Server::send_cors_headers();
			echo \wp_json_encode(
				array(
					'code'    => 'activitypub_proxy_unsafe_host',
					'message' => \__( 'The remote eventStream host is not allowed.', 'activitypub' ),
				)
			);
			exit;
		}

		$context = stream_context_create(
			array(
				'ssl' => array(
					'verify_peer'      => true,
					'verify_peer_name' => true,
					// Pin SNI / cert verification to the original hostname even though we connect by IP.
					'peer_name'        => $host,
				),
			)
		);

		$target = ( false !== strpos( $ip, ':' ) ? '[' . $ip . ']' : $ip ) . ':' . $port;

		// phpcs:ignore WordPress.WP.AlternativeFunctions.file_system_operations_stream_socket_client -- SSE proxy requires raw streaming.
		$stream = stream_socket_client(
			'ssl://' . $target,
			$errno,
			$errstr,
			30,
			STREAM_CLIENT_CONNECT,
			$context
		);

		if ( ! $stream ) {
			\status_header( 502 );
			\header( 'Content-Type: application/json' );
			Server::send_cors_headers();
			echo \wp_json_encode(
				array(
					'code'    => 'activitypub_proxy_connection_failed',
					'message' => \__( 'Failed to connect to the remote eventStream.', 'activitypub' ),
				)
			);
			exit;
		}

		// Send the HTTP request.
		$request_headers  = "GET {$path} HTTP/1.1\r\n";
		$request_headers .= "Host: {$host}\r\n";
		$request_headers .= "Accept: text/event-stream\r\n";
		$request_headers .= "Cache-Control: no-cache\r\n";
		$request_headers .= "Connection: keep-alive\r\n";
		$request_headers .= "\r\n";

		// phpcs:ignore WordPress.WP.AlternativeFunctions.file_system_operations_fwrite -- Raw stream operation.
		fwrite( $stream, $request_headers );

		// Read and skip the HTTP response headers.
		$header_complete = false;
		$status_code     = 0;

		while ( ! feof( $stream ) ) {
			$line = fgets( $stream, 8192 );

			if ( false === $line ) {
				break;
			}

			if ( ! $status_code && preg_match( '/^HTTP\/\d\.\d (\d{3})/', $line, $matches ) ) {
				$status_code = (int) $matches[1];
			}

			// Empty line signals end of headers.
			if ( "\r\n" === $line || "\n" === $line ) {
				$header_complete = true;
				break;
			}
		}

		if ( ! $header_complete || 200 !== $status_code ) {
			// phpcs:ignore WordPress.WP.AlternativeFunctions.file_system_operations_fclose -- Raw stream operation.
			fclose( $stream );
			\status_header( 502 );
			\header( 'Content-Type: application/json' );
			Server::send_cors_headers();
			echo \wp_json_encode(
				array(
					'code'    => 'activitypub_proxy_stream_error',
					'message' => \__( 'The remote eventStream returned an error.', 'activitypub' ),
				)
			);
			exit;
		}

		// Send our own SSE headers and relay the remote stream.
		$this->send_sse_headers();
		$this->send_sse_comment( 'proxying ' . $host );

		$start = time();

		stream_set_timeout( $stream, 10 );

		while ( ! feof( $stream ) && ( time() - $start ) < 300 ) {
			if ( \connection_aborted() ) {
				break;
			}

			$line = fgets( $stream, 8192 );

			if ( false === $line ) {
				$meta = stream_get_meta_data( $stream );

				if ( ! empty( $meta['timed_out'] ) ) {
					$this->send_sse_comment( 'keepalive ' . \gmdate( 'c' ) );
					$this->flush_output();
					continue;
				}

				break;
			}

			// phpcs:ignore WordPress.Security.EscapeOutput.OutputNotEscaped -- Relaying raw SSE protocol data.
			echo $line;
			$this->flush_output();
		}

		// phpcs:ignore WordPress.WP.AlternativeFunctions.file_system_operations_fclose -- Raw stream operation.
		fclose( $stream );

		$this->send_sse_comment( 'proxy timeout' );
		$this->flush_output();

		exit;
	}

	/**
	 * Send SSE-specific HTTP headers.
	 */
	protected function send_sse_headers() {
		while ( ob_get_level() > 0 ) {
			ob_end_clean();
		}

		\status_header( 200 );
		\header( 'Content-Type: text/event-stream' );
		\header( 'Cache-Control: no-cache, no-store' );
		\header( 'Referrer-Policy: no-referrer' );
		\header( 'X-Accel-Buffering: no' );

		// SSE exits before rest_post_dispatch, so CORS must be sent directly.
		Server::send_cors_headers();
	}

	/**
	 * Send an SSE event for a collection item.
	 *
	 * @param \WP_Post $item       The collection post item.
	 * @param string   $collection The collection type ('outbox' or 'inbox').
	 */
	protected function send_sse_event( $item, $collection ) {
		$event_type = $this->get_event_type( $item, $collection );
		$data       = $this->get_event_data( $item, $collection );

		if ( ! $data ) {
			return;
		}

		// phpcs:ignore WordPress.Security.EscapeOutput.OutputNotEscaped -- SSE protocol requires raw output.
		echo 'event: ' . $event_type . "\n";
		// phpcs:ignore WordPress.Security.EscapeOutput.OutputNotEscaped -- SSE protocol requires raw JSON output.
		echo 'data: ' . \wp_json_encode( $data ) . "\n";
		echo 'id: ' . (int) $item->ID . "\n\n";
	}

	/**
	 * Send an SSE comment line.
	 *
	 * @param string $comment The comment text.
	 */
	protected function send_sse_comment( $comment ) {
		// phpcs:ignore WordPress.Security.EscapeOutput.OutputNotEscaped -- SSE protocol requires raw output.
		echo ': ' . $comment . "\n\n";
	}

	/**
	 * Flush all output buffers.
	 */
	protected function flush_output() {
		if ( ob_get_level() > 0 ) {
			ob_flush();
		}
		flush();
	}

	/**
	 * Get the SSE event type for a collection item.
	 *
	 * @param \WP_Post $item       The collection post item.
	 * @param string   $collection The collection type ('outbox' or 'inbox').
	 *
	 * @return string The SSE event type.
	 */
	protected function get_event_type( $item, $collection ) {
		if ( 'inbox' === $collection ) {
			return 'Add';
		}

		$activity_type  = \get_post_meta( $item->ID, '_activitypub_activity_type', true );
		$event_type_map = self::get_event_type_map();

		if ( isset( $event_type_map[ $activity_type ] ) ) {
			return $event_type_map[ $activity_type ];
		}

		return 'Add';
	}

	/**
	 * Get the activity data for a collection item.
	 *
	 * @param \WP_Post $item       The collection post item.
	 * @param string   $collection The collection type ('outbox' or 'inbox').
	 *
	 * @return array|null The activity data, or null on failure.
	 */
	protected function get_event_data( $item, $collection ) {
		if ( 'outbox' === $collection ) {
			$activity = Outbox::get_activity( $item->ID );

			if ( \is_wp_error( $activity ) ) {
				return null;
			}

			return $activity->to_array( false );
		}

		$data = \json_decode( $item->post_content, true );

		return $data ? $data : null;
	}

	/**
	 * Get the latest item ID for a collection.
	 *
	 * @param int    $user_id    The actor ID.
	 * @param string $collection The collection type ('outbox' or 'inbox').
	 *
	 * @return int The latest post ID, or 0 if empty.
	 */
	protected function get_latest_item_id( $user_id, $collection ) {
		$post_type = 'outbox' === $collection ? Outbox::POST_TYPE : Inbox::POST_TYPE;

		$args = array(
			'post_type'      => $post_type,
			'post_status'    => 'any',
			'posts_per_page' => 1,
			'orderby'        => 'ID',
			'order'          => 'DESC',
			'fields'         => 'ids',
			'no_found_rows'  => true,
		);

		if ( 'outbox' === $collection ) {
			// phpcs:ignore WordPress.DB.SlowDBQuery.slow_db_query_meta_query
			$args['meta_query'] = array(
				array(
					'key'   => '_activitypub_activity_actor',
					'value' => Actors::get_type_by_id( $user_id ),
				),
			);
		} else {
			// phpcs:ignore WordPress.DB.SlowDBQuery.slow_db_query_meta_query
			$args['meta_query'] = array(
				array(
					'key'   => '_activitypub_user_id',
					'value' => $user_id,
				),
			);
		}

		$query = new \WP_Query( $args );

		return ! empty( $query->posts ) ? (int) $query->posts[0] : 0;
	}

	/**
	 * Get the eventStream URL for a collection.
	 *
	 * @param int    $user_id    The actor ID.
	 * @param string $collection The collection type ('outbox' or 'inbox').
	 *
	 * @return string The eventStream URL.
	 */
	public function get_stream_url( $user_id, $collection ) {
		return \rest_url( sprintf( '%s/actors/%d/%s/stream', $this->namespace, $user_id, $collection ) );
	}

	/**
	 * Get new collection items since a given ID.
	 *
	 * @param int    $user_id    The actor ID.
	 * @param string $collection The collection type ('outbox' or 'inbox').
	 * @param int    $since_id   Only return items with ID greater than this.
	 *
	 * @return \WP_Post[] Array of new post items.
	 */
	protected function get_new_items( $user_id, $collection, $since_id ) {
		$post_type = 'outbox' === $collection ? Outbox::POST_TYPE : Inbox::POST_TYPE;

		$args = array(
			'post_type'      => $post_type,
			'post_status'    => 'any',
			'posts_per_page' => 20,
			'orderby'        => 'ID',
			'order'          => 'ASC',
			'no_found_rows'  => true,
		);

		if ( 'outbox' === $collection ) {
			// phpcs:ignore WordPress.DB.SlowDBQuery.slow_db_query_meta_query
			$args['meta_query'] = array(
				array(
					'key'   => '_activitypub_activity_actor',
					'value' => Actors::get_type_by_id( $user_id ),
				),
			);
		} else {
			// phpcs:ignore WordPress.DB.SlowDBQuery.slow_db_query_meta_query
			$args['meta_query'] = array(
				array(
					'key'   => '_activitypub_user_id',
					'value' => $user_id,
				),
			);
		}

		if ( $since_id > 0 ) {
			$where_filter = function ( $where ) use ( $since_id ) {
				global $wpdb;
				$where .= $wpdb->prepare( " AND {$wpdb->posts}.ID > %d", $since_id );
				return $where;
			};
			\add_filter( 'posts_where', $where_filter );
		}

		$query = new \WP_Query( $args );

		if ( $since_id > 0 ) {
			\remove_filter( 'posts_where', $where_filter );
		}

		return $query->posts;
	}
}