<?php
/**
 * Background ingestion orchestration for DataStream.
 *
 * Since 1.4.0 the engine syncs a set of streams (each one external API).
 * Installs that never configured streams fall back to the 1.3-era single-API
 * options, presented internally as one legacy stream.
 *
 * @package DataStream
 */

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

/**
 * Class DataStream_Engine
 */
class DataStream_Engine {

	/**
	 * Singleton instance.
	 *
	 * @var self|null
	 */
	private static $instance = null;

	/**
	 * DAO instance.
	 *
	 * @var DataStream_DAO
	 */
	private $dao;

	/**
	 * JSON decode depth limit.
	 */
	private const JSON_DEPTH = 32;

	/**
	 * Retrieve the shared engine instance.
	 *
	 * @return self
	 */
	public static function get_instance() {
		if ( null === self::$instance ) {
			self::$instance = new self();
		}

		return self::$instance;
	}

	/**
	 * Constructor.
	 */
	private function __construct() {
		$this->dao = new DataStream_DAO();
		add_action( DATASTREAM_CRON_HOOK, array( $this, 'run_ingestion_sync' ) );
	}

	/**
	 * Schedule the recurring ingestion event.
	 */
	public static function schedule_sync() {
		if ( ! DataStream_Config::get( 'cron_enabled' ) ) {
			self::unschedule_sync();
			return;
		}

		$schedule = DataStream_Config::validate_schedule( (string) DataStream_Config::get( 'cron_schedule' ) );

		if ( ! wp_next_scheduled( DATASTREAM_CRON_HOOK ) ) {
			wp_schedule_event( time() + 60, $schedule, DATASTREAM_CRON_HOOK );
		}
	}

	/**
	 * Reschedule cron when the interval changes.
	 *
	 * @param string $schedule Cron schedule slug.
	 */
	public static function reschedule_sync( $schedule ) {
		self::unschedule_sync();

		if ( ! DataStream_Config::get( 'cron_enabled' ) ) {
			return;
		}

		$scheduled = wp_schedule_event( time() + 60, DataStream_Config::validate_schedule( $schedule ), DATASTREAM_CRON_HOOK );

		if ( false === $scheduled ) {
			DataStream_Logger::error( 'Failed to schedule ingestion cron event.' );
		}
	}

	/**
	 * Remove all scheduled ingestion events.
	 */
	public static function unschedule_sync() {
		$timestamp = wp_next_scheduled( DATASTREAM_CRON_HOOK );

		while ( $timestamp ) {
			wp_unschedule_event( $timestamp, DATASTREAM_CRON_HOOK );
			$timestamp = wp_next_scheduled( DATASTREAM_CRON_HOOK );
		}
	}

	/**
	 * Sync all due streams (or one specific stream).
	 *
	 * @param bool   $force       Bypass the sync lock and backoff (manual sync).
	 * @param string $only_stream Sync only this stream id.
	 * @return array<string, mixed> Aggregated result; per-stream detail under 'streams'.
	 */
	public function run_ingestion_sync( $force = false, $only_stream = '' ) {
		$started_at = microtime( true );
		$streams    = $this->resolve_streams( $only_stream );

		if ( empty( $streams ) ) {
			return array(
				'success'   => false,
				'processed' => 0,
				'skipped'   => 0,
				'message'   => __( 'No streams configured. Add one under DataStream → Settings → Streams.', 'datastream-api-sync' ),
			);
		}

		// One stream fully in backoff and nothing else to do: skip before
		// taking the lock, so the skip is cheap and leaves no trace.
		if ( ! $force ) {
			$due = array();

			foreach ( $streams as $stream ) {
				if ( ! $this->stream_backing_off( $stream ) ) {
					$due[] = $stream;
				}
			}

			if ( empty( $due ) ) {
				DataStream_Logger::info( 'Sync skipped: all streams backing off after repeated failures.' );

				return array(
					'success'   => false,
					'processed' => 0,
					'skipped'   => 0,
					'message'   => __( 'Sync skipped: backing off after repeated failures.', 'datastream-api-sync' ),
					'backoff'   => true,
				);
			}

			$streams = $due;
		}

		if ( ! $force && ! DataStream_Sync_Lock::acquire() ) {
			$result = array(
				'success'   => false,
				'processed' => 0,
				'skipped'   => 0,
				'message'   => __( 'Sync skipped: another ingestion is already running.', 'datastream-api-sync' ),
				'locked'    => true,
			);
			DataStream_Config::record_sync_result( $result );
			return $result;
		}

		if ( $force ) {
			DataStream_Sync_Lock::force_release();
			DataStream_Sync_Lock::acquire();
		}

		$aggregate = array(
			'success'   => true,
			'processed' => 0,
			'skipped'   => 0,
			'failed'    => 0,
			'streams'   => array(),
		);
		$messages  = array();

		foreach ( $streams as $stream ) {
			try {
				$result = $this->execute_stream_sync( $stream );
			} catch ( Exception $exception ) {
				DataStream_Logger::error(
					'Unexpected ingestion exception.',
					array(
						'stream'  => (string) $stream['id'],
						'message' => $exception->getMessage(),
					)
				);

				$result = array(
					'success'   => false,
					'processed' => 0,
					'skipped'   => 0,
					'message'   => __( 'Unexpected ingestion error.', 'datastream-api-sync' ),
				);
			}

			$this->update_stream_backoff( $stream, ! empty( $result['success'] ), $result['retry_after'] ?? 0 );

			if ( empty( $stream['_legacy'] ) ) {
				DataStream_Streams::record_stream_result(
					(string) $stream['id'],
					! empty( $result['success'] ),
					(string) ( $result['message'] ?? '' ),
					(int) ( $result['processed'] ?? 0 )
				);
			}

			$aggregate['success']    = $aggregate['success'] && ! empty( $result['success'] );
			$aggregate['processed'] += (int) ( $result['processed'] ?? 0 );
			$aggregate['skipped']   += (int) ( $result['skipped'] ?? 0 );
			$aggregate['failed']    += (int) ( $result['failed'] ?? 0 );

			$aggregate['streams'][ (string) $stream['id'] ] = $result;
			$messages[] = $stream['id'] . ': ' . (string) ( $result['message'] ?? '' );
		}

		$aggregate['message']     = implode( ' | ', $messages );
		$aggregate['duration_ms'] = (int) round( ( microtime( true ) - $started_at ) * 1000 );

		DataStream_Config::record_sync_result( $aggregate );
		DataStream_Sync_Lock::release();

		return $aggregate;
	}

	/**
	 * Whether scheduled syncs are currently paused by the failure backoff
	 * (true only when every configured stream is backing off).
	 *
	 * @return bool
	 */
	public function is_backing_off() {
		$streams = $this->resolve_streams( '' );

		if ( empty( $streams ) ) {
			return false;
		}

		foreach ( $streams as $stream ) {
			if ( ! $this->stream_backing_off( $stream ) ) {
				return false;
			}
		}

		return true;
	}

	/**
	 * The streams this run should consider.
	 *
	 * @param string $only_stream Restrict to one stream id.
	 * @return array<int, array<string, mixed>>
	 */
	private function resolve_streams( $only_stream ) {
		if ( '' !== $only_stream ) {
			$stream = DataStream_Streams::get_stream( $only_stream );
			return $stream ? array( $stream ) : array();
		}

		$streams = array_values( DataStream_Streams::get_active_streams() );

		if ( ! empty( $streams ) ) {
			return $streams;
		}

		$legacy = $this->legacy_stream();

		return $legacy ? array( $legacy ) : array();
	}

	/**
	 * Present the 1.3-era single-API options as a stream.
	 *
	 * @return array<string, mixed>|null
	 */
	private function legacy_stream() {
		$api_url = esc_url_raw( (string) DataStream_Config::get( 'api_url' ) );

		if ( '' === $api_url ) {
			return null;
		}

		return wp_parse_args(
			array(
				'id'               => sanitize_key( (string) DataStream_Config::get( 'stream_name' ) ) ?: '',
				'_legacy'          => true,
				'api_url'          => $api_url,
				'api_method'       => (string) DataStream_Config::get( 'api_method' ),
				'api_headers'      => (string) DataStream_Config::get( 'api_headers' ),
				'api_bearer_token' => (string) DataStream_Config::get( 'api_bearer_token' ),
				'json_items_path'  => (string) DataStream_Config::get( 'json_items_path' ),
				'id_field'         => (string) DataStream_Config::get( 'id_field' ),
				'content_field'    => (string) DataStream_Config::get( 'content_field' ),
			),
			DataStream_Streams::defaults()
		);
	}

	/**
	 * Whether a stream is inside its failure-backoff window.
	 *
	 * @param array<string, mixed> $stream Stream config.
	 * @return bool
	 */
	private function stream_backing_off( $stream ) {
		if ( ! empty( $stream['_legacy'] ) ) {
			return time() < (int) DataStream_Config::get( 'backoff_until' );
		}

		return time() < (int) ( $stream['backoff_until'] ?? 0 );
	}

	/**
	 * Track consecutive failures per stream and set an exponential backoff
	 * window (15m doubling, capped at 6h), so a dead endpoint is not hammered
	 * every interval forever. A 429 Retry-After hint overrides the window.
	 *
	 * @param array<string, mixed> $stream      Stream config.
	 * @param bool                 $success     Whether the sync succeeded.
	 * @param int                  $retry_after Server-requested delay in seconds (0 = none).
	 */
	private function update_stream_backoff( $stream, $success, $retry_after = 0 ) {
		$legacy   = ! empty( $stream['_legacy'] );
		$failures = $legacy ? absint( DataStream_Config::get( 'failure_count' ) ) : absint( $stream['failure_count'] ?? 0 );

		if ( $success ) {
			if ( $failures > 0 || ( $legacy ? (int) DataStream_Config::get( 'backoff_until' ) : (int) ( $stream['backoff_until'] ?? 0 ) ) > 0 ) {
				$this->persist_stream_backoff( $stream, 0, 0 );
			}
			return;
		}

		++$failures;
		$until = $legacy ? (int) DataStream_Config::get( 'backoff_until' ) : (int) ( $stream['backoff_until'] ?? 0 );

		if ( $retry_after > 0 ) {
			// The API told us when to come back; believe it (capped at 1h).
			$until = time() + min( HOUR_IN_SECONDS, $retry_after );
		} elseif ( $failures >= 3 ) {
			$delay = min( 6 * HOUR_IN_SECONDS, 15 * MINUTE_IN_SECONDS * ( 2 ** min( 10, $failures - 3 ) ) );
			$until = time() + $delay;

			DataStream_Logger::error(
				'Repeated sync failures — backing off.',
				array(
					'stream'               => (string) $stream['id'],
					'consecutive_failures' => $failures,
					'retry_after_seconds'  => $delay,
				)
			);
		}

		$this->persist_stream_backoff( $stream, $failures, $until );
	}

	/**
	 * Write backoff state to wherever this stream keeps it.
	 *
	 * @param array<string, mixed> $stream   Stream config.
	 * @param int                  $failures Consecutive failures.
	 * @param int                  $until    Unix timestamp, 0 = clear.
	 */
	private function persist_stream_backoff( $stream, $failures, $until ) {
		if ( ! empty( $stream['_legacy'] ) ) {
			DataStream_Config::set( 'failure_count', $failures );
			DataStream_Config::set( 'backoff_until', $until );
			return;
		}

		DataStream_Streams::update_runtime_state( (string) $stream['id'], $failures, $until );
	}

	/**
	 * Sync one stream: fetch (paginated), map, persist, optionally map to posts.
	 *
	 * @param array<string, mixed> $stream Stream config.
	 * @return array<string, mixed>
	 */
	private function execute_stream_sync( $stream ) {
		$fetch = $this->fetch_stream_items( $stream );

		if ( ! $fetch['success'] ) {
			return $fetch;
		}

		$items   = $fetch['items'];
		$records = array();
		$skipped = 0;
		$max     = absint( DataStream_Config::get( 'max_records_per_sync' ) );

		foreach ( $items as $item ) {
			if ( $max > 0 && count( $records ) >= $max ) {
				++$skipped;
				continue;
			}

			$mapped = $this->map_item( $item, $stream );

			if ( null === $mapped ) {
				++$skipped;
				continue;
			}

			$records[] = $mapped;
		}

		if ( empty( $records ) ) {
			$this->maybe_purge_stale();

			DataStream_Logger::info(
				'Ingestion completed with zero mappable records.',
				array(
					'stream'  => (string) $stream['id'],
					'skipped' => $skipped,
				)
			);

			return array(
				'success'   => true,
				'processed' => 0,
				'skipped'   => $skipped,
				'message'   => __( 'No records to process.', 'datastream-api-sync' ),
			);
		}

		$batch = $this->dao->batch_upsert( $records );

		$posts_mapped = 0;

		if ( $batch['success'] ) {
			foreach ( $records as $record ) {
				DataStream_Provider::invalidate_cache( $record['external_id'] );
			}

			if ( '' !== (string) ( $stream['post_type'] ?? '' ) && class_exists( 'DataStream_Post_Mapper' ) ) {
				$posts_mapped = DataStream_Post_Mapper::sync_records( $stream, $records );
				DataStream_Post_Mapper::handle_stale_posts( $stream );
			}

			$this->maybe_purge_stale();

			DataStream_Logger::info(
				'Ingestion batch completed.',
				array(
					'stream'    => (string) $stream['id'],
					'processed' => $batch['processed'],
					'skipped'   => $skipped + (int) $batch['skipped'],
					'failed'    => (int) $batch['failed'],
					'posts'     => $posts_mapped,
				)
			);
		}

		if ( $batch['success'] ) {
			$message = sprintf(
				/* translators: 1: records stored, 2: records skipped */
				__( 'Processed %1$d records (%2$d skipped).', 'datastream-api-sync' ),
				(int) $batch['processed'],
				$skipped + (int) $batch['skipped']
			);

			if ( (int) $batch['failed'] > 0 ) {
				$message .= ' ' . sprintf(
					/* translators: %d: number of failed records */
					__( '%d records failed and were skipped.', 'datastream-api-sync' ),
					(int) $batch['failed']
				);
			}
		} else {
			$message = __( 'Batch ingestion failed.', 'datastream-api-sync' );
		}

		if ( $posts_mapped > 0 ) {
			$message .= ' ' . sprintf(
				/* translators: %d: number of posts created or updated */
				__( '%d posts synced.', 'datastream-api-sync' ),
				$posts_mapped
			);
		}

		return array(
			'success'      => $batch['success'],
			'processed'    => (int) $batch['processed'],
			'skipped'      => $skipped + (int) $batch['skipped'],
			'failed'       => (int) $batch['failed'],
			'posts_mapped' => $posts_mapped,
			'message'      => $message,
		);
	}

	/**
	 * Fetch every page of a stream and return normalized items.
	 *
	 * @param array<string, mixed> $stream Stream config.
	 * @return array<string, mixed> success, items|message, retry_after.
	 */
	private function fetch_stream_items( $stream ) {
		$mode      = (string) ( $stream['pagination'] ?? 'none' );
		$max_pages = 'none' === $mode ? 1 : max( 1, min( DataStream_Feature_Gate::max_pages(), absint( $stream['max_pages'] ) ) );
		$items     = array();
		$last_sig  = null;

		for ( $page = 1; $page <= $max_pages; $page++ ) {
			$url      = DataStream_Streams::request_url( $stream, $page );
			$response = $this->remote_request( $url, $stream );

			if ( is_wp_error( $response ) ) {
				DataStream_Logger::error(
					'Remote API request failed.',
					array(
						'stream' => (string) $stream['id'],
						'error'  => $response->get_error_code(),
					)
				);

				return array(
					'success'   => false,
					'processed' => 0,
					'skipped'   => 0,
					'message'   => __( 'Remote API request failed.', 'datastream-api-sync' ),
					'error'     => $response->get_error_code(),
				);
			}

			$status_code = (int) wp_remote_retrieve_response_code( $response );

			if ( 429 === $status_code ) {
				$retry_after = absint( wp_remote_retrieve_header( $response, 'retry-after' ) );

				DataStream_Logger::error(
					'Remote API rate limited the sync (HTTP 429).',
					array(
						'stream'      => (string) $stream['id'],
						'retry_after' => $retry_after,
					)
				);

				return array(
					'success'     => false,
					'processed'   => 0,
					'skipped'     => 0,
					'message'     => __( 'Rate limited by the API (HTTP 429). Sync will retry later.', 'datastream-api-sync' ),
					'status'      => 429,
					'retry_after' => $retry_after > 0 ? $retry_after : 15 * MINUTE_IN_SECONDS,
				);
			}

			if ( $status_code < 200 || $status_code >= 300 ) {
				DataStream_Logger::error(
					'Remote API returned non-success status.',
					array(
						'stream' => (string) $stream['id'],
						'status' => $status_code,
					)
				);

				return array(
					'success'   => false,
					'processed' => 0,
					'skipped'   => 0,
					'message'   => sprintf(
						/* translators: %d: HTTP status code */
						__( 'Remote API returned HTTP %d.', 'datastream-api-sync' ),
						$status_code
					),
					'status'    => $status_code,
				);
			}

			$decoded = $this->decode_body( $response, (string) $stream['id'], (string) ( $stream['format'] ?? 'auto' ) );

			if ( null === $decoded ) {
				return array(
					'success'   => false,
					'processed' => 0,
					'skipped'   => 0,
					'message'   => __( 'The response could not be read as JSON, XML, RSS, or CSV.', 'datastream-api-sync' ),
				);
			}

			$page_items = $this->normalize_items( $decoded, $stream );

			if ( empty( $page_items ) ) {
				break;
			}

			// Fickle-API guard: some endpoints ignore the page parameter and
			// return the same page forever. Stop when a page repeats.
			$sig = md5( (string) wp_json_encode( $page_items[0] ) . count( $page_items ) );

			if ( $sig === $last_sig ) {
				DataStream_Logger::info(
					'Pagination stopped: API returned an identical page.',
					array(
						'stream' => (string) $stream['id'],
						'page'   => $page,
					)
				);
				break;
			}

			$last_sig = $sig;
			$items    = array_merge( $items, $page_items );

			if ( 'none' === $mode ) {
				break;
			}
		}

		return array(
			'success' => true,
			'items'   => $items,
		);
	}

	/**
	 * Decode a response body through the format parser (JSON, XML, RSS/Atom,
	 * CSV — BOM handled, big JSON integers preserved as strings).
	 *
	 * @param array<string, mixed> $response  HTTP response.
	 * @param string               $stream_id For logging.
	 * @param string               $format    Stream format setting ('auto' sniffs).
	 * @return array<string, mixed>|null
	 */
	private function decode_body( $response, $stream_id, $format = 'auto' ) {
		$body = trim( (string) wp_remote_retrieve_body( $response ) );

		if ( '' === $body ) {
			DataStream_Logger::error( 'Remote API returned an empty body.', array( 'stream' => $stream_id ) );
			return null;
		}

		$content_type = wp_remote_retrieve_header( $response, 'content-type' );
		$content_type = is_array( $content_type ) ? (string) reset( $content_type ) : (string) $content_type;

		$parsed = DataStream_Parser::parse( $body, $content_type, $format );

		if ( null === $parsed ) {
			$cap = DataStream_Config::max_response_bytes();

			DataStream_Logger::error(
				'Remote response could not be parsed as JSON, XML, RSS, or CSV.',
				array(
					'stream'          => $stream_id,
					'body_bytes'      => strlen( $body ),
					'maybe_truncated' => strlen( $body ) >= $cap,
				)
			);

			return null;
		}

		return $parsed['data'];
	}

	/**
	 * Test API connectivity without persisting data.
	 *
	 * @param string $url       Optional override URL.
	 * @param string $stream_id Optional stream whose auth/mapping to use.
	 * @return array<string, mixed>
	 */
	public function test_connection( $url = '', $stream_id = '' ) {
		$stream = '' !== $stream_id ? DataStream_Streams::get_stream( $stream_id ) : null;

		if ( null === $stream ) {
			$stream = $this->legacy_stream();
		}

		if ( null === $stream ) {
			$stream = wp_parse_args( array( '_legacy' => true ), DataStream_Streams::defaults() );
		}

		if ( '' !== $url ) {
			$stream['api_url'] = esc_url_raw( $url );
		}

		$api_url = (string) $stream['api_url'];

		if ( '' === $api_url ) {
			return array(
				'success' => false,
				'message' => __( 'API URL is empty.', 'datastream-api-sync' ),
			);
		}

		$started  = microtime( true );
		$response = $this->remote_request( DataStream_Streams::request_url( $stream, 1 ), $stream );
		$elapsed  = (int) round( ( microtime( true ) - $started ) * 1000 );

		if ( is_wp_error( $response ) ) {
			return array(
				'success'     => false,
				'message'     => $response->get_error_message(),
				'error_code'  => $response->get_error_code(),
				'duration_ms' => $elapsed,
			);
		}

		$status = (int) wp_remote_retrieve_response_code( $response );
		$body   = trim( (string) wp_remote_retrieve_body( $response ) );
		$data   = '' !== $body ? $this->decode_body( $response, (string) $stream['id'], (string) ( $stream['format'] ?? 'auto' ) ) : array();
		$items  = is_array( $data ) ? $this->normalize_items( $data, $stream ) : array();

		return array(
			'success'     => $status >= 200 && $status < 300,
			'message'     => $status >= 200 && $status < 300
				? __( 'Connection successful.', 'datastream-api-sync' )
				: sprintf(
					/* translators: %d: HTTP status code */
					__( 'HTTP %d received.', 'datastream-api-sync' ),
					$status
				),
			'status'      => $status,
			'duration_ms' => $elapsed,
			'body_bytes'  => strlen( $body ),
			'item_count'  => count( $items ),
			'json_valid'  => is_array( $data ),
			'sample'      => $items[0] ?? null,
		);
	}

	/**
	 * Perform the HTTP request for a stream, with one retry on transient
	 * failure (network error or 5xx) — fickle APIs often need exactly one.
	 *
	 * @param string                    $url    Endpoint URL (already paginated/keyed).
	 * @param array<string, mixed>|null $stream Stream config (null = legacy globals).
	 * @return array<string, mixed>|WP_Error
	 */
	private function remote_request( $url, $stream = null ) {
		if ( null === $stream ) {
			$stream = $this->legacy_stream();

			if ( null === $stream ) {
				return new WP_Error( 'no_stream', __( 'No stream configuration available.', 'datastream-api-sync' ) );
			}
		}

		$headers = DataStream_Streams::request_headers( $stream );

		// OAuth-configured streams (premium) fetch/refresh their token here;
		// a failed token request leaves the call unauthenticated, and the
		// resulting 401 flows into the normal per-stream backoff.
		//
		// class_exists comes FIRST: the free build strips the OAuth file
		// entirely, and calling a static on a missing class is a fatal --
		// it would fire on every stream fetch, not just OAuth ones.
		if ( class_exists( 'DataStream_OAuth' ) && DataStream_Feature_Gate::is_premium() && DataStream_OAuth::is_configured( $stream ) ) {
			$token = DataStream_OAuth::bearer_for( $stream );

			if ( null !== $token ) {
				$headers['Authorization'] = 'Bearer ' . $token;
			}
		}

		$args = array(
			'timeout'             => DataStream_Config::sanitize_timeout( DataStream_Config::get( 'api_timeout' ) ),
			'redirection'         => 3,
			'sslverify'           => (bool) DataStream_Config::get( 'api_ssl_verify' ),
			'headers'             => $headers,
			// Core's wp_http_validate_url() blocks localhost and private/reserved
			// IP ranges (it resolves hostnames too). Opt out for intranet APIs via
			// the DATASTREAM_ALLOW_PRIVATE_URLS constant or the filter.
			'reject_unsafe_urls'  => DataStream_Config::reject_unsafe_urls(),
			'limit_response_size' => DataStream_Config::max_response_bytes(),
		);

		$method   = DataStream_Config::sanitize_api_method( (string) ( $stream['api_method'] ?? 'GET' ) );
		$response = $this->dispatch_request( $method, $url, $args );

		if ( $this->is_transient_failure( $response ) ) {
			$delay = absint( apply_filters( 'datastream_retry_delay_seconds', 2 ) );

			DataStream_Logger::info(
				'Transient API failure — retrying once.',
				array(
					'stream' => (string) ( $stream['id'] ?? '' ),
					'delay'  => $delay,
				)
			);

			if ( $delay > 0 ) {
				sleep( min( 10, $delay ) );
			}

			$response = $this->dispatch_request( $method, $url, $args );
		}

		return $response;
	}

	/**
	 * One HTTP dispatch.
	 *
	 * @param string               $method GET|POST.
	 * @param string               $url    URL.
	 * @param array<string, mixed> $args   Request args.
	 * @return array<string, mixed>|WP_Error
	 */
	private function dispatch_request( $method, $url, $args ) {
		if ( 'POST' === $method ) {
			return wp_remote_post( $url, $args );
		}

		return wp_remote_get( $url, $args );
	}

	/**
	 * Whether a response warrants the single retry.
	 *
	 * @param array<string, mixed>|WP_Error $response Response.
	 * @return bool
	 */
	private function is_transient_failure( $response ) {
		if ( is_wp_error( $response ) ) {
			return true;
		}

		$status = (int) wp_remote_retrieve_response_code( $response );

		return $status >= 500 && $status < 600;
	}

	/**
	 * Normalize API response into a flat list of items.
	 *
	 * @internal Public for the test suite; not a stable API.
	 *
	 * @param array<string, mixed>      $data   Decoded API response.
	 * @param array<string, mixed>|null $stream Stream config (null = legacy globals).
	 * @return array<int, array<string, mixed>>
	 */
	public function normalize_items( $data, $stream = null ) {
		$path = null !== $stream
			? trim( (string) ( $stream['json_items_path'] ?? '' ) )
			: trim( (string) DataStream_Config::get( 'json_items_path' ) );

		if ( '' !== $path ) {
			$resolved = $this->resolve_json_path( $data, $path );

			if ( is_array( $resolved ) ) {
				return $this->flatten_items( $resolved );
			}
		}

		foreach ( array( 'data', 'items', 'results', 'records', 'entries' ) as $key ) {
			if ( isset( $data[ $key ] ) && is_array( $data[ $key ] ) ) {
				return $this->flatten_items( $data[ $key ] );
			}
		}

		return $this->flatten_items( $data );
	}

	/**
	 * Resolve dot-notation JSON path.
	 *
	 * Segments are matched verbatim (minus surrounding whitespace) — APIs use
	 * camelCase keys like `dataItems`, and lowercasing them here would make
	 * those paths silently unreachable.
	 *
	 * @internal Public for the test suite; not a stable API.
	 *
	 * @param array<string, mixed> $data Root data.
	 * @param string               $path Dot path (e.g. response.dataItems).
	 * @return mixed
	 */
	public function resolve_json_path( $data, $path ) {
		$segments = array_filter( array_map( 'trim', explode( '.', $path ) ), 'strlen' );
		$current  = $data;

		foreach ( $segments as $segment ) {
			if ( ! is_array( $current ) || ! array_key_exists( $segment, $current ) ) {
				return null;
			}

			$current = $current[ $segment ];
		}

		return $current;
	}

	/**
	 * Flatten mixed list/object payloads.
	 *
	 * @param mixed $data Candidate items.
	 * @return array<int, array<string, mixed>>
	 */
	private function flatten_items( $data ) {
		if ( ! is_array( $data ) ) {
			return array();
		}

		if ( $this->is_list_array( $data ) ) {
			$items = array();

			foreach ( $data as $item ) {
				if ( is_array( $item ) ) {
					$items[] = $item;
				}
			}

			return $items;
		}

		return array( $data );
	}

	/**
	 * Determine whether an array is a sequential list.
	 *
	 * @param array<mixed> $array Input array.
	 * @return bool
	 */
	private function is_list_array( $array ) {
		if ( array() === $array ) {
			return true;
		}

		return array_keys( $array ) === range( 0, count( $array ) - 1 );
	}

	/**
	 * Map a single API item to a persistence record.
	 *
	 * @internal Public for the test suite; not a stable API.
	 *
	 * @param array<string, mixed>      $item   Single API record.
	 * @param array<string, mixed>|null $stream Stream config (null = legacy globals).
	 * @return array{external_id: string, payload: string, stream_type: string, source: string}|null
	 */
	public function map_item( $item, $stream = null ) {
		if ( ! is_array( $item ) ) {
			return null;
		}

		// Field names are matched verbatim — sanitize_key() would lowercase
		// `userId` into `userid` and the lookup would quietly find nothing.
		if ( null !== $stream ) {
			$id_field      = trim( (string) ( $stream['id_field'] ?? '' ) ) ?: 'id';
			$content_field = trim( (string) ( $stream['content_field'] ?? '' ) );
			$override      = empty( $stream['_legacy'] )
				? (string) $stream['id']
				: sanitize_key( (string) DataStream_Config::get( 'stream_name' ) );
		} else {
			$id_field      = trim( (string) DataStream_Config::get( 'id_field' ) ) ?: 'id';
			$content_field = trim( (string) DataStream_Config::get( 'content_field' ) );
			$override      = sanitize_key( (string) DataStream_Config::get( 'stream_name' ) );
		}

		if ( null !== $stream && 'hash' === (string) ( $stream['id_source'] ?? 'field' ) ) {
			// APIs with no unique field: fingerprint the record itself. The
			// same content maps to the same row; changed content is a new
			// record (stale-record purge cleans the old ones up).
			$external_id = 'h_' . md5( (string) wp_json_encode( $item ) );
		} else {
			$raw_id = $item[ $id_field ] ?? ( $item['id'] ?? ( $item['external_id'] ?? ( $item['uuid'] ?? null ) ) );

			if ( null === $raw_id || ( is_string( $raw_id ) && '' === trim( $raw_id ) ) ) {
				return null;
			}

			if ( is_array( $raw_id ) || is_object( $raw_id ) ) {
				return null;
			}

			$external_id = $this->normalize_external_id( $raw_id );

			if ( '' === $external_id ) {
				return null;
			}
		}

		// Only an explicitly configured content field narrows the payload.
		// (1.2.0 also treated a key literally named "content" as the whole
		// payload — which silently discarded every other field on any API
		// that happens to have one. RSS feeds always do.)
		if ( '' !== $content_field && isset( $item[ $content_field ] ) ) {
			$payload_data = $item[ $content_field ];
		} else {
			$payload_data = $item;
		}

		if ( ! is_array( $payload_data ) && ! is_scalar( $payload_data ) ) {
			return null;
		}

		if ( ! is_array( $payload_data ) ) {
			$payload_data = array( 'value' => $payload_data );
		}

		$payload = wp_json_encode( $this->sanitize_payload( $payload_data ) );

		if ( false === $payload || 'null' === $payload ) {
			DataStream_Logger::error(
				'Failed to encode payload.',
				array( 'external_id' => $external_id )
			);
			return null;
		}

		$meta = DataStream_Chronicles::resolve_stream_meta( $item, $override );

		return array(
			'external_id' => $external_id,
			'payload'     => $payload,
			'stream_type' => $meta['stream_type'],
			'source'      => $meta['source'],
		);
	}

	/**
	 * Normalize external IDs with strict typing rules.
	 *
	 * Numerics are cast to string without intval(): truncating floats or
	 * clamping large IDs would map distinct API records onto one another.
	 * (Genuinely big integer IDs never reach here as floats — the JSON decode
	 * uses JSON_BIGINT_AS_STRING, so they arrive as strings.)
	 *
	 * @internal Public for the test suite; not a stable API.
	 *
	 * @param mixed $id Raw identifier.
	 * @return string
	 */
	public function normalize_external_id( $id ) {
		if ( is_bool( $id ) ) {
			return $id ? '1' : '0';
		}

		if ( is_int( $id ) || is_float( $id ) ) {
			return (string) $id;
		}

		$id = sanitize_text_field( (string) $id );

		return trim( $id );
	}

	/**
	 * Recursively sanitize payload values before persistence.
	 *
	 * Stores the API's data faithfully: keys keep their exact case and
	 * characters (an API returning `firstName` must be readable as
	 * `firstName` later), and string values keep newlines and markup.
	 * The only rewriting done here is what the storage layer requires:
	 * valid UTF-8, no NUL bytes, no control characters in keys.
	 * Escaping happens on output — the renderer and REST layer own that.
	 *
	 * @internal Public for the test suite; not a stable API.
	 *
	 * @param mixed $value Raw value from API.
	 * @return mixed
	 */
	public function sanitize_payload( $value ) {
		if ( is_array( $value ) ) {
			$sanitized = array();

			foreach ( $value as $key => $item ) {
				$sanitized[ $this->sanitize_payload_key( $key ) ] = $this->sanitize_payload( $item );
			}

			return $sanitized;
		}

		if ( is_int( $value ) || is_float( $value ) || is_bool( $value ) ) {
			return $value;
		}

		if ( is_string( $value ) ) {
			return $this->sanitize_payload_string( $value );
		}

		return null;
	}

	/**
	 * Sanitize an array key: preserve case and content, strip control chars.
	 *
	 * @param int|string $key Raw key.
	 * @return int|string
	 */
	private function sanitize_payload_key( $key ) {
		if ( is_int( $key ) ) {
			return $key;
		}

		$key = $this->sanitize_payload_string( (string) $key );

		return (string) preg_replace( '/[\x00-\x1F\x7F]/', '', $key );
	}

	/**
	 * Make a string safe to JSON-encode without altering its content:
	 * enforce valid UTF-8 and remove NUL bytes, nothing else.
	 *
	 * @param string $value Raw string.
	 * @return string
	 */
	private function sanitize_payload_string( $value ) {
		if ( function_exists( 'mb_check_encoding' ) ) {
			$valid = mb_check_encoding( $value, 'UTF-8' );
		} else {
			// PCRE rejects invalid UTF-8 subjects, so an empty /u match doubles
			// as a validity check on hosts without mbstring.
			$valid = ( 1 === preg_match( '//u', $value ) );
		}

		if ( ! $valid ) {
			if ( function_exists( 'mb_convert_encoding' ) ) {
				$value = (string) mb_convert_encoding( $value, 'UTF-8', 'UTF-8' );
			} else {
				$value = (string) preg_replace( '/[^\x09\x0A\x0D\x20-\x7E]/', '', $value );
			}
		}

		return str_replace( "\0", '', $value );
	}

	/**
	 * Purge stale records when configured.
	 */
	private function maybe_purge_stale() {
		$days = absint( DataStream_Config::get( 'stale_record_days' ) );

		if ( $days < 1 ) {
			return;
		}

		$deleted = $this->dao->purge_stale_records( $days );

		if ( $deleted > 0 ) {
			DataStream_Logger::info(
				'Purged stale records.',
				array(
					'deleted' => $deleted,
					'days'    => $days,
				)
			);
		}
	}
}
