<?php
/** Incremental SSE decoding using the WordPress HTTP API's cURL transport. */
namespace AISupportAssistant\Providers;
if ( ! defined( 'ABSPATH' ) ) { exit; }

class StreamTransport {
	private string $buffer = '';
	private string $answer = '';
	private array $event_data = array();
	private bool $finished = false;
	private string $provider;
	private $on_delta;

	public function __construct( string $provider, callable $on_delta ) {
		$this->provider = $provider;
		$this->on_delta = $on_delta;
	}

	/** Decode arbitrary transport boundaries, including split UTF-8 and CRLF. */
	public function feed( string $bytes ): void {
		$this->buffer .= $bytes;
		while ( false !== ( $end = strpos( $this->buffer, "\n" ) ) ) {
			$line = rtrim( substr( $this->buffer, 0, $end ), "\r" );
			$this->buffer = substr( $this->buffer, $end + 1 );
			if ( 0 === strpos( $line, 'data:' ) ) {
				$this->event_data[] = ltrim( substr( $line, 5 ), ' ' );
				continue;
			}
			if ( '' !== $line || ! $this->event_data ) { continue; }
			$data = trim( implode( "\n", $this->event_data ) );
			$this->event_data = array();
			if ( '[DONE]' === $data ) { $this->finished = true; continue; }
			$event = json_decode( $data, true );
			if ( ! is_array( $event ) ) { throw new \RuntimeException( 'Invalid provider stream data.' ); }
			if ( isset( $event['error'] ) ) {
				$code = is_array( $event['error'] ) ? (int) ( $event['error']['code'] ?? 0 ) : 0;
				$error_message = esc_html( self::error_message( $event, sanitize_text_field( $this->provider ) ) );
				// phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- The exception message is escaped immediately above.
				throw new \RuntimeException( $error_message, (int) $code );
			}
			$text = '';
			if ( 'openai' === $this->provider ) {
				$text = $event['choices'][0]['delta']['content'] ?? '';
			} else {
				foreach ( $event['candidates'][0]['content']['parts'] ?? array() as $part ) {
					if ( empty( $part['thought'] ) ) { $text .= $part['text'] ?? ''; }
				}
				if ( ! empty( $event['candidates'][0]['finishReason'] ) ) { $this->finished = true; }
			}
			if ( is_string( $text ) && '' !== $text ) {
				$this->answer .= $text;
				( $this->on_delta )( $text );
			}
		}
		if ( strlen( $this->buffer ) + array_sum( array_map( 'strlen', $this->event_data ) ) > 1048576 ) { throw new \RuntimeException( 'Provider stream frame exceeds the supported size.' ); }
	}

	public function answer(): string {
		if ( ! $this->finished || '' === trim( $this->answer ) ) { throw new \RuntimeException( 'The provider stream ended before completing an answer.' ); }
		return trim( $this->answer );
	}

	/** Keep provider diagnostics for admin logs; public REST errors remain generic. */
	private static function error_message( array $payload, string $provider, int $status = 0 ): string {
		$error = $payload['error'] ?? array();
		$message = is_array( $error ) && is_string( $error['message'] ?? null ) ? $error['message'] : 'The provider rejected the streaming request.';
		$code = is_array( $error ) ? ( $error['code'] ?? $status ) : $status;
		$code = is_scalar( $code ) ? sanitize_text_field( (string) $code ) : (string) (int) $status;
		return ucfirst( sanitize_text_field( $provider ) ) . ' streaming error' . ( '' !== $code && '0' !== $code ? ' (' . $code . ')' : '' ) . ': ' . substr( $message, 0, 2000 );
	}

	public static function post( string $url, array $args, string $provider, callable $on_delta ): string {
		return self::retry( static function ( $emit ) use ( $url, $args, $provider ) {
			return self::request_once( $url, $args, $provider, $emit );
		}, $on_delta );
	}

	/** Retry transient failures only before any content has been delivered. */
	public static function retry( callable $request, callable $on_delta ): string {
		$emitted = false;
		$emit = static function ( $text ) use ( $on_delta, &$emitted ) {
			$emitted = true;
			$on_delta( $text );
		};
		for ( $attempt = 0; ; ++$attempt ) {
			try { return $request( $emit ); }
			catch ( \RuntimeException $error ) {
				if ( $emitted || $attempt >= 2 || ! in_array( $error->getCode(), array( 502, 503, 504 ), true ) ) { throw $error; }
				usleep( 500000 * ( $attempt + 1 ) );
			}
		}
	}

	private static function request_once( string $url, array $args, string $provider, callable $on_delta ): string {
		$decoder = new self( $provider, $on_delta );
		$streamed = false;
		$stream_error = null;
		$error_body = '';
		$hook = static function ( $handle, $request_args, $request_url ) use ( $url, $decoder, &$streamed, &$stream_error, &$error_body ) {
			if ( $request_url !== $url ) { return; }
			$streamed = true;
			// phpcs:ignore WordPress.WP.AlternativeFunctions.curl_curl_setopt -- Configure only WordPress's existing transport handle for incremental delivery; wp_safe_remote_post retains request, TLS, and proxy control.
			curl_setopt( $handle, CURLOPT_WRITEFUNCTION, static function ( $curl, $bytes ) use ( $decoder, &$stream_error, &$error_body ) {
				// phpcs:ignore WordPress.WP.AlternativeFunctions.curl_curl_getinfo -- Inspect the active WordPress transport status before emitting bytes; response helpers are available only after streaming completes.
				if ( ! $stream_error && 200 === (int) curl_getinfo( $curl, CURLINFO_HTTP_CODE ) ) {
					// Consume the remaining body on decode failure: Requests retries cURL write errors.
					try { $decoder->feed( $bytes ); } catch ( \Throwable $error ) { $stream_error = $error; }
				} elseif ( strlen( $error_body ) < 65536 ) {
					$error_body .= substr( $bytes, 0, 65536 - strlen( $error_body ) );
				}
				return strlen( $bytes );
			} );
		};
		add_action( 'http_api_curl', $hook, 10, 3 );
		try {
			$args['redirection'] = 0;
			$response = wp_safe_remote_post( $url, $args );
		} finally { remove_action( 'http_api_curl', $hook, 10 ); }
		if ( $stream_error ) { throw $stream_error; }
		if ( is_wp_error( $response ) ) {
			throw new \RuntimeException( esc_html( ucfirst( $provider ) . ' streaming connection failed: ' . $response->get_error_message() ) );
		}
		$status = wp_remote_retrieve_response_code( $response );
		if ( 200 !== $status ) {
			$payload = json_decode( $streamed ? $error_body : wp_remote_retrieve_body( $response ), true );
			$error_message = esc_html( self::error_message( is_array( $payload ) ? $payload : array(), sanitize_text_field( $provider ), (int) $status ) );
			// phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- The exception message is escaped immediately above.
			throw new \RuntimeException( $error_message, (int) $status );
		}
		// Hosts without cURL still support the protocol, with a buffered HTTP transport.
		if ( ! $streamed ) { $decoder->feed( wp_remote_retrieve_body( $response ) ); }
		$decoder->feed( "\n\n" );
		return $decoder->answer();
	}
}