<?php
/**
 * Stream registry: each stream is one external API with its own connection,
 * auth, mapping, pagination, and post-mapping configuration.
 *
 * Stored in a single option. The 1.3-era single-API options migrate into a
 * stream named after the configured stream name (or "default") on upgrade.
 *
 * @package DataStream
 */

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

/**
 * Class DataStream_Streams
 */
class DataStream_Streams {

	/**
	 * Option holding all stream configs.
	 */
	public const OPTION = 'datastream_streams';

	/**
	 * A blank stream with every supported key.
	 *
	 * @return array<string, mixed>
	 */
	public static function defaults() {
		return array(
			'id'                 => '',
			'label'              => '',
			'enabled'            => true,
			// Connection.
			'api_url'            => '',
			'api_method'         => 'GET',
			'api_headers'        => '',
			'api_bearer_token'   => '',
			'api_basic_user'     => '',
			'api_basic_pass'     => '',
			'api_key_param'      => '',
			'api_key_value'      => '',
			// Mapping.
			'format'             => 'auto', // auto | json | xml | rss | csv.
			'json_items_path'    => '',
			'id_field'           => 'id',
			'id_source'          => 'field', // field | hash (fingerprint records with no unique field).
			'content_field'      => '',
			// Pagination.
			'pagination'         => 'none', // none | page | offset.
			'page_param'         => 'page',
			'per_page_param'     => '',
			'per_page'           => 100,
			'max_pages'          => 1,
			// API to Posts (premium).
			'post_type'          => '',
			'post_title_field'   => '',
			'post_content_field' => '',
			'post_status'        => 'publish',
			'sideload_images'    => false,
			'image_field'        => '',
			'taxonomy_map'       => '', // JSON {"taxonomy":"field.path"}.
			'meta_map'           => '', // JSON {"meta_key":"field.path"}.
			'price_field'        => '', // WooCommerce product preset.
			'sku_field'          => '',
			'stock_field'        => '',
			'stale_post_action'  => 'keep', // keep | draft | trash.
			// OAuth 2.0 (premium).
			'oauth_token_url'    => '',
			'oauth_client_id'    => '',
			'oauth_client_secret' => '',
			'oauth_scope'        => '',
			'oauth_refresh_token' => '',
			// Runtime state (managed by the engine, not the UI).
			'failure_count'      => 0,
			'backoff_until'      => 0,
			'last_sync_at'       => 0,
			'last_success'       => true,
			'last_message'       => '',
			'last_processed'     => 0,
			'history'            => array(), // Last 10 runs, 1 = success.
		);
	}

	/**
	 * All streams, keyed by id.
	 *
	 * @return array<string, array<string, mixed>>
	 */
	public static function get_streams() {
		$stored = get_option( self::OPTION, array() );

		if ( ! is_array( $stored ) ) {
			$stored = array();
		}

		$streams = array();

		foreach ( $stored as $stream ) {
			if ( ! is_array( $stream ) || empty( $stream['id'] ) ) {
				continue;
			}

			$streams[ $stream['id'] ] = wp_parse_args( $stream, self::defaults() );
		}

		return $streams;
	}

	/**
	 * Streams the engine should sync now: enabled, with a URL, capped by tier.
	 *
	 * @return array<string, array<string, mixed>>
	 */
	public static function get_active_streams() {
		$active = array();
		$cap    = DataStream_Feature_Gate::max_enabled_streams();

		foreach ( self::get_streams() as $id => $stream ) {
			if ( count( $active ) >= $cap ) {
				break;
			}

			if ( ! empty( $stream['enabled'] ) && '' !== (string) $stream['api_url'] ) {
				$active[ $id ] = $stream;
			}
		}

		return $active;
	}

	/**
	 * One stream by id.
	 *
	 * @param string $id Stream id.
	 * @return array<string, mixed>|null
	 */
	public static function get_stream( $id ) {
		$streams = self::get_streams();
		return $streams[ sanitize_key( (string) $id ) ] ?? null;
	}

	/**
	 * Create or update a stream. Input is sanitized here — this is the only
	 * write path (admin UI, CLI, and migration all come through it).
	 *
	 * @param array<string, mixed> $stream Stream config (id required).
	 * @return array<string, mixed>|WP_Error Saved stream or error.
	 */
	public static function save_stream( $stream ) {
		if ( ! is_array( $stream ) ) {
			return new WP_Error( 'invalid_stream', __( 'Stream must be an array.', 'datastream-api-sync' ) );
		}

		$id = sanitize_key( (string) ( $stream['id'] ?? '' ) );

		if ( '' === $id ) {
			return new WP_Error( 'invalid_stream_id', __( 'Stream id is required (lowercase letters, numbers, underscores).', 'datastream-api-sync' ) );
		}

		$streams  = self::get_streams();
		$existing = $streams[ $id ] ?? self::defaults();
		$clean    = self::sanitize_stream( $stream, $existing );

		$clean['id'] = $id;
		$streams[ $id ] = $clean;

		update_option( self::OPTION, array_values( $streams ), false );

		return $clean;
	}

	/**
	 * Delete a stream (records in the ingestion table are kept; purge them
	 * separately if wanted).
	 *
	 * @param string $id Stream id.
	 * @return bool Whether a stream was removed.
	 */
	public static function delete_stream( $id ) {
		$id      = sanitize_key( (string) $id );
		$streams = self::get_streams();

		if ( ! isset( $streams[ $id ] ) ) {
			return false;
		}

		unset( $streams[ $id ] );
		update_option( self::OPTION, array_values( $streams ), false );

		return true;
	}

	/**
	 * Update only the runtime backoff state of a stream.
	 *
	 * @param string $id            Stream id.
	 * @param int    $failure_count Consecutive failures.
	 * @param int    $backoff_until Unix timestamp, 0 = none.
	 */
	public static function update_runtime_state( $id, $failure_count, $backoff_until ) {
		$streams = self::get_streams();
		$id      = sanitize_key( (string) $id );

		if ( ! isset( $streams[ $id ] ) ) {
			return;
		}

		$streams[ $id ]['failure_count'] = absint( $failure_count );
		$streams[ $id ]['backoff_until'] = absint( $backoff_until );

		update_option( self::OPTION, array_values( $streams ), false );
	}

	/**
	 * Record the outcome of one sync run for the dashboard cards.
	 *
	 * @param string $id        Stream id.
	 * @param bool   $success   Whether the run succeeded.
	 * @param string $message   Human-readable outcome.
	 * @param int    $processed Records stored.
	 */
	public static function record_stream_result( $id, $success, $message, $processed ) {
		$streams = self::get_streams();
		$id      = sanitize_key( (string) $id );

		if ( ! isset( $streams[ $id ] ) ) {
			return;
		}

		$history   = is_array( $streams[ $id ]['history'] ) ? $streams[ $id ]['history'] : array();
		$history[] = $success ? 1 : 0;

		$streams[ $id ]['last_sync_at']   = time();
		$streams[ $id ]['last_success']   = (bool) $success;
		$streams[ $id ]['last_message']   = sanitize_text_field( (string) $message );
		$streams[ $id ]['last_processed'] = absint( $processed );
		$streams[ $id ]['history']        = array_slice( $history, -10 );

		update_option( self::OPTION, array_values( $streams ), false );
	}

	/**
	 * Sanitize a stream config.
	 *
	 * @param array<string, mixed> $input    Raw values.
	 * @param array<string, mixed> $existing Current values (kept where input omits a key).
	 * @return array<string, mixed>
	 */
	public static function sanitize_stream( $input, $existing ) {
		$defaults = self::defaults();
		$existing = wp_parse_args( $existing, $defaults );
		$clean    = $existing;

		if ( array_key_exists( 'label', $input ) ) {
			$clean['label'] = sanitize_text_field( (string) $input['label'] );
		}

		if ( array_key_exists( 'enabled', $input ) ) {
			$clean['enabled'] = ! empty( $input['enabled'] );
		}

		if ( array_key_exists( 'api_url', $input ) ) {
			$clean['api_url'] = DataStream_Config::sanitize_api_url( (string) $input['api_url'] );
		}

		if ( array_key_exists( 'api_method', $input ) ) {
			$clean['api_method'] = DataStream_Config::sanitize_api_method( (string) $input['api_method'] );
		}

		if ( array_key_exists( 'api_headers', $input ) ) {
			$clean['api_headers'] = DataStream_Config::sanitize_headers_json( (string) $input['api_headers'] );
		}

		// Secrets: blank input means "keep existing" so forms can re-save
		// without re-entering them; the literal string "-" clears.
		foreach ( array( 'api_bearer_token', 'api_basic_pass', 'oauth_client_secret', 'oauth_refresh_token' ) as $secret ) {
			if ( array_key_exists( $secret, $input ) ) {
				$value = trim( (string) $input[ $secret ] );
				if ( '-' === $value ) {
					$clean[ $secret ] = '';
				} elseif ( '' !== $value && '********' !== $value ) {
					$clean[ $secret ] = sanitize_text_field( $value );
				}
			}
		}

		foreach ( array( 'api_basic_user', 'api_key_param', 'api_key_value', 'json_items_path', 'id_field', 'content_field', 'page_param', 'per_page_param', 'post_title_field', 'post_content_field', 'image_field', 'price_field', 'sku_field', 'stock_field', 'oauth_client_id', 'oauth_scope' ) as $text ) {
			if ( array_key_exists( $text, $input ) ) {
				$clean[ $text ] = trim( sanitize_text_field( (string) $input[ $text ] ) );
			}
		}

		if ( array_key_exists( 'oauth_token_url', $input ) ) {
			$clean['oauth_token_url'] = '' === trim( (string) $input['oauth_token_url'] )
				? ''
				: DataStream_Config::sanitize_api_url( (string) $input['oauth_token_url'] );
		}

		if ( array_key_exists( 'sideload_images', $input ) ) {
			$clean['sideload_images'] = ! empty( $input['sideload_images'] );
		}

		if ( array_key_exists( 'stale_post_action', $input ) ) {
			$action                     = sanitize_key( (string) $input['stale_post_action'] );
			$clean['stale_post_action'] = in_array( $action, array( 'keep', 'draft', 'trash' ), true ) ? $action : 'keep';
		}

		// Mapping JSONs: object of string => string, or empty.
		foreach ( array( 'taxonomy_map', 'meta_map' ) as $map_key ) {
			if ( ! array_key_exists( $map_key, $input ) ) {
				continue;
			}

			$raw = trim( (string) $input[ $map_key ] );

			if ( '' === $raw ) {
				$clean[ $map_key ] = '';
				continue;
			}

			$decoded = json_decode( $raw, true );
			$safe    = array();

			if ( is_array( $decoded ) ) {
				foreach ( $decoded as $target => $source ) {
					if ( is_string( $target ) && is_string( $source ) && '' !== trim( $target ) && '' !== trim( $source ) ) {
						$safe[ sanitize_text_field( trim( $target ) ) ] = sanitize_text_field( trim( $source ) );
					}
				}
			}

			$clean[ $map_key ] = empty( $safe ) ? '' : (string) wp_json_encode( $safe );
		}

		if ( '' === $clean['id_field'] ) {
			$clean['id_field'] = 'id';
		}

		if ( array_key_exists( 'id_source', $input ) ) {
			$source             = sanitize_key( (string) $input['id_source'] );
			$clean['id_source'] = in_array( $source, array( 'field', 'hash' ), true ) ? $source : 'field';
		}

		if ( array_key_exists( 'format', $input ) ) {
			$format          = sanitize_key( (string) $input['format'] );
			$clean['format'] = in_array( $format, DataStream_Parser::FORMATS, true ) ? $format : 'auto';
		}

		if ( array_key_exists( 'pagination', $input ) ) {
			$mode                = sanitize_key( (string) $input['pagination'] );
			$clean['pagination'] = in_array( $mode, array( 'none', 'page', 'offset' ), true ) ? $mode : 'none';
		}

		if ( array_key_exists( 'per_page', $input ) ) {
			$clean['per_page'] = DataStream_Config::sanitize_bounded_int( $input['per_page'], 1, 1000, 100 );
		}

		if ( array_key_exists( 'max_pages', $input ) ) {
			$clean['max_pages'] = DataStream_Config::sanitize_bounded_int( $input['max_pages'], 1, DataStream_Feature_Gate::max_pages(), 1 );
		}

		if ( array_key_exists( 'post_type', $input ) ) {
			$clean['post_type'] = sanitize_key( (string) $input['post_type'] );
		}

		if ( array_key_exists( 'post_status', $input ) ) {
			$status               = sanitize_key( (string) $input['post_status'] );
			$clean['post_status'] = in_array( $status, array( 'publish', 'draft', 'pending', 'private' ), true ) ? $status : 'publish';
		}

		return $clean;
	}

	/**
	 * Build request headers for a stream, honoring wp-config constants.
	 *
	 * Constant override order: DATASTREAM_TOKEN_{STREAMID} (uppercased) wins,
	 * then DATASTREAM_API_BEARER_TOKEN (legacy, applies to every stream that
	 * has no per-stream constant), then the stored token.
	 *
	 * @param array<string, mixed> $stream Stream config.
	 * @return array<string, string>
	 */
	public static function request_headers( $stream ) {
		$json = (string) ( $stream['api_headers'] ?? '' );

		if ( '' === $json ) {
			$headers = array();
		} else {
			$decoded = json_decode( $json, true );
			$headers = is_array( $decoded ) ? $decoded : array();
		}

		$constant = 'DATASTREAM_TOKEN_' . strtoupper( (string) ( $stream['id'] ?? '' ) );

		if ( defined( $constant ) && '' !== (string) constant( $constant ) ) {
			$token = (string) constant( $constant );
		} elseif ( ! empty( $stream['_legacy'] ) && defined( 'DATASTREAM_API_BEARER_TOKEN' ) && '' !== (string) DATASTREAM_API_BEARER_TOKEN ) {
			// The 1.3-era global constant applies ONLY to the legacy
			// single-API setup. Applying it to every named stream would send
			// one API's credential to all the others.
			$token = (string) DATASTREAM_API_BEARER_TOKEN;
		} else {
			$token = (string) ( $stream['api_bearer_token'] ?? '' );
		}

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

		$user = (string) ( $stream['api_basic_user'] ?? '' );
		$pass = (string) ( $stream['api_basic_pass'] ?? '' );

		if ( '' !== $user && ! isset( $headers['Authorization'] ) ) {
			// phpcs:ignore WordPress.PHP.DiscouragedPHPFunctions.obfuscation_base64_encode -- Basic auth is defined as base64.
			$headers['Authorization'] = 'Basic ' . base64_encode( $user . ':' . $pass );
		}

		return $headers;
	}

	/**
	 * Build the request URL for one page of a stream.
	 *
	 * @param array<string, mixed> $stream Stream config.
	 * @param int                  $page   1-based page number.
	 * @return string
	 */
	public static function request_url( $stream, $page = 1 ) {
		$url  = (string) ( $stream['api_url'] ?? '' );
		$args = array();

		$key_param = (string) ( $stream['api_key_param'] ?? '' );
		$key_value = (string) ( $stream['api_key_value'] ?? '' );

		if ( '' !== $key_param && '' !== $key_value ) {
			$args[ $key_param ] = $key_value;
		}

		$mode = (string) ( $stream['pagination'] ?? 'none' );

		if ( 'page' === $mode ) {
			$args[ (string) $stream['page_param'] ?: 'page' ] = $page;

			if ( '' !== (string) $stream['per_page_param'] ) {
				$args[ (string) $stream['per_page_param'] ] = absint( $stream['per_page'] );
			}
		} elseif ( 'offset' === $mode ) {
			$per_page = max( 1, absint( $stream['per_page'] ) );
			$args[ (string) $stream['page_param'] ?: 'offset' ] = ( $page - 1 ) * $per_page;

			if ( '' !== (string) $stream['per_page_param'] ) {
				$args[ (string) $stream['per_page_param'] ] = $per_page;
			}
		}

		if ( empty( $args ) ) {
			return $url;
		}

		// WP's add_query_arg() does not urlencode values; encode them here so
		// API keys with special characters survive.
		return add_query_arg( array_map( 'rawurlencode', array_map( 'strval', $args ) ), $url );
	}

	/**
	 * Migrate the 1.3-era single-API options into a stream, once.
	 */
	public static function migrate_legacy() {
		$existing = get_option( self::OPTION, null );

		if ( is_array( $existing ) && ! empty( $existing ) ) {
			return;
		}

		$legacy_url = (string) get_option( 'datastream_api_url', '' );

		if ( '' === $legacy_url ) {
			return;
		}

		$id = sanitize_key( (string) get_option( 'datastream_stream_name', '' ) ) ?: 'default';

		self::save_stream(
			array(
				'id'               => $id,
				'label'            => ucfirst( $id ),
				'enabled'          => true,
				'api_url'          => $legacy_url,
				'api_method'       => (string) get_option( 'datastream_api_method', 'GET' ),
				'api_headers'      => (string) get_option( 'datastream_api_headers', '' ),
				'api_bearer_token' => (string) get_option( 'datastream_api_bearer_token', '' ),
				'json_items_path'  => (string) get_option( 'datastream_json_items_path', '' ),
				'id_field'         => (string) get_option( 'datastream_id_field', 'id' ),
				'content_field'    => (string) get_option( 'datastream_content_field', '' ),
			)
		);

		DataStream_Logger::info( 'Migrated legacy single-API configuration into stream.', array( 'stream' => $id ) );
	}
}
