<?php
/**
 * Phase 5: low-level helpers for streaming LLM responses.
 *
 * WordPress' wp_remote_post() buffers the entire HTTP response in memory
 * before returning, so we drop to raw cURL with a CURLOPT_WRITEFUNCTION
 * callback to dispatch each incoming chunk to the client as it arrives.
 *
 * Two pieces here:
 *   1. atml_curl_stream_post() — POSTs to an HTTPS URL and invokes a
 *      callback for each received body chunk. Caller decides what to do
 *      with each chunk (typically pipe it to the browser as SSE).
 *   2. atml_parse_sse_buffer() — incremental SSE parser. Accumulates raw
 *      HTTP bytes across multiple WRITEFUNCTION ticks, yields fully-formed
 *      `event:` + `data:` records once each one is complete (terminated
 *      by a blank line).
 *
 * Why pure cURL and not wp_safe_remote_post(): WP_Http_Curl doesn't expose
 * a streaming write callback. We could fork it, but a 30-line direct
 * cURL call is cleaner and keeps the streaming path isolated.
 *
 * @package Antimanual
 * @since   3.7.0
 */

if ( ! defined( 'ABSPATH' ) ) {
	exit;
}

/**
 * Server-sent-events frame writer.
 *
 * Emits one event in the canonical SSE format and flushes the output buffer
 * so the byte hits the wire immediately. The caller is responsible for
 * having already disabled output compression / nginx buffering via the
 * usual headers (Content-Type: text/event-stream, X-Accel-Buffering: no).
 *
 * @param string $event Event name (e.g. 'chunk', 'meta', 'done', 'error').
 * @param mixed  $data  JSON-serializable payload.
 */
function atml_sse_emit( string $event, $data ): void {
	echo 'event: ' . $event . "\n";
	echo 'data: ' . wp_json_encode( $data ) . "\n\n";
	// Two-stage flush: drain PHP's output buffer stack, then ask the SAPI
	// to send what's in its own buffer. On nginx + PHP-FPM with proxy_buffering
	// off this hits the browser in <5 ms.
	if ( function_exists( 'ob_get_level' ) ) {
		while ( ob_get_level() > 0 ) {
			@ob_end_flush();
		}
	}
	@flush();
}

/**
 * Prepare the current request for SSE streaming.
 *
 * Disables WordPress' standard output handling, sends streaming-friendly
 * headers, and signals the SAPI to write each event byte to the wire
 * immediately. Idempotent within a single request.
 *
 * Reverse-proxy notes:
 *   - nginx: requires `proxy_buffering off` in the location, or the
 *     `X-Accel-Buffering: no` header we send here (nginx honours it).
 *   - Apache + mod_deflate: gzip-compresses SSE streams unless disabled.
 *     We send `Content-Encoding: identity` to defeat it.
 *   - Cloudflare: requires "Compression" off OR a 200ms keep-alive ping
 *     so the proxy flushes the buffer. We send a ` ` comment every few
 *     seconds via atml_sse_ping().
 */
function atml_sse_open(): void {
	if ( ! headers_sent() ) {
		nocache_headers();
		header( 'Content-Type: text/event-stream; charset=utf-8' );
		header( 'Cache-Control: no-cache, no-store, must-revalidate, private' );
		header( 'X-Accel-Buffering: no' );        // nginx
		header( 'Content-Encoding: identity' );    // defeat mod_deflate
		header( 'Connection: keep-alive' );
	}
	// Disable PHP's output buffer.
	if ( function_exists( 'ob_get_level' ) ) {
		while ( ob_get_level() > 0 ) {
			@ob_end_clean();
		}
	}
	// Implicit flush so any echo within this request hits the wire.
	@ini_set( 'zlib.output_compression', '0' );
	@ini_set( 'output_buffering', '0' );
	@ini_set( 'implicit_flush', '1' );
	if ( function_exists( 'ob_implicit_flush' ) ) {
		ob_implicit_flush( true );
	}
}

/**
 * Send an SSE comment line. Used as a keep-alive to prevent intermediate
 * proxies (Cloudflare, corporate firewalls) from closing the connection
 * during long token generations.
 */
function atml_sse_ping(): void {
	echo ": ping\n\n";
	if ( function_exists( 'ob_get_level' ) ) {
		while ( ob_get_level() > 0 ) {
			@ob_end_flush();
		}
	}
	@flush();
}

/**
 * POST a JSON body and dispatch each received chunk to a callback.
 *
 * @param string   $url       Endpoint URL.
 * @param array    $headers   ['Header-Name: value', …]. Authorization etc.
 * @param string   $body_json Request body (already JSON-encoded).
 * @param callable $on_chunk  fn(string $bytes): void — called with each chunk
 *                            of raw HTTP body as it arrives.
 * @param int      $timeout   Hard timeout in seconds for the upstream call.
 * @return array{ok: bool, status: int, error?: string}
 */
function atml_curl_stream_post( string $url, array $headers, string $body_json, callable $on_chunk, int $timeout = 180 ): array {
	if ( ! function_exists( 'curl_init' ) ) {
		return [ 'ok' => false, 'status' => 0, 'error' => 'cURL extension is required for streaming.' ];
	}

	$ch = curl_init( $url );
	if ( ! $ch ) {
		return [ 'ok' => false, 'status' => 0, 'error' => 'Failed to init cURL.' ];
	}

	$status = 0;
	curl_setopt_array( $ch, [
		CURLOPT_POST           => true,
		CURLOPT_HTTPHEADER     => array_values( $headers ),
		CURLOPT_POSTFIELDS     => $body_json,
		CURLOPT_CONNECTTIMEOUT => 10,
		CURLOPT_TIMEOUT        => max( 30, $timeout ),
		CURLOPT_USERAGENT      => 'antimanual-php/' . ( defined( 'ANTIMANUAL_VERSION' ) ? ANTIMANUAL_VERSION : '0' ),
		// Don't gather the whole body — pipe chunks straight to the caller.
		CURLOPT_RETURNTRANSFER => false,
		CURLOPT_WRITEFUNCTION  => static function ( $ch_inner, $bytes ) use ( $on_chunk ) {
			if ( '' !== $bytes ) {
				$on_chunk( $bytes );
			}
			return strlen( $bytes );
		},
		// Capture HTTP status from the response headers but don't buffer them.
		CURLOPT_HEADERFUNCTION => static function ( $ch_inner, $header ) use ( &$status ) {
			if ( 0 === stripos( $header, 'HTTP/' ) ) {
				$parts  = preg_split( '/\s+/', trim( $header ), 3 );
				$status = isset( $parts[1] ) ? (int) $parts[1] : 0;
			}
			return strlen( $header );
		},
	] );

	$ok    = curl_exec( $ch );
	$errno = curl_errno( $ch );
	$error = curl_error( $ch );
	curl_close( $ch );

	if ( false === $ok || 0 !== $errno ) {
		return [
			'ok'     => false,
			'status' => $status,
			'error'  => $error ?: ( 'cURL errno ' . $errno ),
		];
	}

	if ( $status >= 400 ) {
		return [
			'ok'     => false,
			'status' => $status,
			'error'  => sprintf( 'Upstream returned HTTP %d.', $status ),
		];
	}

	return [ 'ok' => true, 'status' => $status ];
}

/**
 * Incremental SSE parser.
 *
 * SSE events are separated by a blank line; each event is a series of
 * `field: value` lines. Multiple `data:` lines accumulate. We're only
 * interested in `event:` and `data:` here — `id:` / `retry:` are ignored.
 *
 * @param string   $bytes       Raw bytes received in this tick.
 * @param string   $buffer      Carryover buffer (caller maintains, passed by ref).
 * @param callable $on_event    fn(string $event_name, string $data_text): void
 *                              — called for each fully-parsed event.
 */
function atml_parse_sse_buffer( string $bytes, string &$buffer, callable $on_event ): void {
	$buffer .= $bytes;

	while ( ( $pos = strpos( $buffer, "\n\n" ) ) !== false ) {
		$frame  = substr( $buffer, 0, $pos );
		$buffer = substr( $buffer, $pos + 2 );

		$event_name = 'message';
		$data_lines = [];

		foreach ( explode( "\n", $frame ) as $line ) {
			$line = rtrim( $line, "\r" );
			if ( '' === $line || ':' === $line[0] ) {
				// blank or comment — skip
				continue;
			}
			$colon = strpos( $line, ':' );
			if ( false === $colon ) {
				continue;
			}
			$field = substr( $line, 0, $colon );
			$value = ltrim( substr( $line, $colon + 1 ) );

			if ( 'event' === $field ) {
				$event_name = $value;
			} elseif ( 'data' === $field ) {
				$data_lines[] = $value;
			}
		}

		$data_text = implode( "\n", $data_lines );
		$on_event( $event_name, $data_text );
	}
}
