<?php
/**
 * Serialized now-playing worker.
 *
 * All metadata fetching funnels through here: a queue of station ids (option) drained by a
 * single worker at a time (global lock), one station after another, within a time budget.
 * The worker is started by the recurring poll (every minute) and kicked on demand through a
 * non-blocking loopback request whenever a reader finds its cache cold.
 *
 * @package StreamSera
 */

namespace StreamSera\Jobs;

use StreamSera\Metadata\Resolver;
use StreamSera\PostTypes\Station;

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

/**
 * Background job that polls now-playing metadata for every station.
 */
class Metadata_Poll {

	const HOOK         = 'streamsera_poll_metadata';
	const QUEUE_OPTION = 'streamsera_np_queue';
	const LOCK         = 'streamsera_np_worker_lock';
	const KICK_LOCK    = 'streamsera_np_worker_kick';
	const AJAX_ACTION  = 'streamsera_np_worker';

	/**
	 * Transient that throttles the schedule check on `init`.
	 */
	const SCHEDULE_CHECK = 'streamsera_poll_scheduled';

	/**
	 * Seconds one worker run may spend before handing over to the next run.
	 *
	 * Kept well inside PHP's default 30 s max_execution_time (one station refresh can still be in
	 * flight when the budget runs out), so the worker never needs to change PHP limits.
	 */
	const BUDGET = 12;

	/**
	 * Hook in.
	 */
	public function register() {
		add_action( self::HOOK, array( $this, 'run' ) );
		add_action( 'init', array( $this, 'schedule' ), 20 );
		add_action( 'update_option_streamsera_settings', array( $this, 'reset_schedule_check' ) );
		// Station published, unpublished or deleted: re-check whether the poll is needed.
		add_action( 'transition_post_status', array( $this, 'maybe_reset_on_station_change' ), 10, 3 );
		add_action( 'deleted_post', array( $this, 'reset_schedule_check' ) );
		add_action( 'wp_ajax_nopriv_' . self::AJAX_ACTION, array( $this, 'ajax_worker' ) );
		add_action( 'wp_ajax_' . self::AJAX_ACTION, array( $this, 'ajax_worker' ) );
	}

	/**
	 * Schedule the recurring poll (every minute; the cache TTL decides actual fetch frequency).
	 */
	public function schedule() {
		// Checked at most every 10 minutes (or right after settings / station changes).
		if ( get_transient( self::SCHEDULE_CHECK ) ) {
			return;
		}
		set_transient( self::SCHEDULE_CHECK, 1, 10 * MINUTE_IN_SECONDS );
		if ( ! streamsera_get_setting( 'background_poll', true ) || ! self::has_published_stations() ) {
			// Turned off, or nothing to poll yet: make sure no recurring poll is left behind.
			Scheduler::cancel( self::HOOK );
			wp_clear_scheduled_hook( self::HOOK );
			return;
		}
		Scheduler::ensure_recurring( self::HOOK, 60 );
	}

	/**
	 * Whether at least one station is published (cheap: one id).
	 *
	 * @return bool
	 */
	private static function has_published_stations() {
		$ids = get_posts(
			array(
				'post_type'        => Station::POST_TYPE,
				'post_status'      => 'publish',
				'numberposts'      => 1,
				'fields'           => 'ids',
				'no_found_rows'    => true,
				'suppress_filters' => false,
			)
		);
		return ! empty( $ids );
	}

	/**
	 * A station changed status: re-check the schedule on the next request.
	 *
	 * @param string   $new_status New status.
	 * @param string   $old_status Old status.
	 * @param \WP_Post $post       Post.
	 */
	public function maybe_reset_on_station_change( $new_status, $old_status, $post ) {
		if ( $post instanceof \WP_Post && Station::POST_TYPE === $post->post_type && ( 'publish' === $new_status || 'publish' === $old_status ) ) {
			$this->reset_schedule_check();
		}
	}

	/**
	 * Re-check the schedule on the next request after settings change (e.g. polling re-enabled).
	 */
	public function reset_schedule_check() {
		delete_transient( self::SCHEDULE_CHECK );
	}

	/**
	 * Recurring poll: queue every published station whose cache is cold, then drain.
	 */
	public function run() {
		if ( ! streamsera_get_setting( 'background_poll', true ) ) {
			return;
		}
		$ids = get_posts(
			array(
				'post_type'   => Station::POST_TYPE,
				'post_status' => 'publish',
				'numberposts' => -1,
				'fields'      => 'ids',
			)
		);
		foreach ( $ids as $id ) {
			if ( ! Resolver::is_fresh( $id ) ) {
				self::push( (int) $id );
			}
		}
		self::drain();
		/**
		 * Fires after the background metadata poll.
		 *
		 * @param array $ids Station IDs considered.
		 */
		do_action( 'streamsera_after_poll', $ids );
	}

	/* ---------------- queue ---------------- */

	/**
	 * Queue one station and make sure a worker will pick it up.
	 *
	 * @param int $station_id Station.
	 */
	public static function enqueue( $station_id ) {
		if ( self::push( $station_id ) ) {
			self::kick();
		}
	}

	/**
	 * Add a station to the queue (deduped).
	 *
	 * @param int $station_id Station.
	 * @return bool Whether the queue changed.
	 */
	private static function push( $station_id ) {
		$station_id = absint( $station_id );
		if ( ! $station_id ) {
			return false;
		}
		$queue = self::queue();
		if ( in_array( $station_id, $queue, true ) ) {
			return false;
		}
		$queue[] = $station_id;
		update_option( self::QUEUE_OPTION, $queue, false );
		return true;
	}

	/**
	 * Current queue (fresh read; the option is not autoloaded and may be changed by other processes).
	 *
	 * @return int[]
	 */
	private static function queue() {
		wp_cache_delete( self::QUEUE_OPTION, 'options' );
		$queue = get_option( self::QUEUE_OPTION, array() );
		return is_array( $queue ) ? array_values( array_unique( array_map( 'absint', $queue ) ) ) : array();
	}

	/**
	 * Pop the next station id.
	 *
	 * @return int 0 when empty.
	 */
	private static function shift() {
		$queue = self::queue();
		if ( empty( $queue ) ) {
			return 0;
		}
		$next = (int) array_shift( $queue );
		update_option( self::QUEUE_OPTION, $queue, false );
		return $next;
	}

	/* ---------------- worker ---------------- */

	/**
	 * Process the queue one station at a time until it is empty or the time budget is spent.
	 * Only one worker runs at a time across all PHP processes.
	 *
	 * @return int Stations refreshed in this run.
	 */
	public static function drain() {
		if ( get_transient( self::LOCK ) ) {
			return 0;
		}
		/**
		 * Filter the time budget (seconds) of one worker run.
		 *
		 * Keep it well below the host's max_execution_time: the plugin does not change PHP limits.
		 *
		 * @param int $budget Seconds.
		 */
		$budget = min( 20, max( 5, (int) apply_filters( 'streamsera_poll_budget', self::BUDGET ) ) );
		// If the process is stopped early, the lock simply expires and the next run picks the queue up.
		set_transient( self::LOCK, time(), $budget + Resolver::LOCK_TTL );

		$deadline = microtime( true ) + $budget;
		$done     = 0;
		try {
			while ( microtime( true ) < $deadline ) {
				$id = self::shift();
				if ( ! $id ) {
					break;
				}
				if ( 'publish' !== get_post_status( $id ) ) {
					continue;
				}
				if ( Resolver::is_fresh( $id ) ) {
					continue;
				}
				Resolver::refresh( $id );
				++$done;
			}
		} finally {
			delete_transient( self::LOCK );
		}

		// Work left over: hand it to a fresh process instead of running on.
		if ( ! empty( self::queue() ) ) {
			self::kick();
		}
		return $done;
	}

	/**
	 * Start a worker in another process without waiting for it (non-blocking loopback).
	 * Throttled so a burst of cold readers triggers one request, not one per reader.
	 */
	public static function kick() {
		if ( get_transient( self::LOCK ) || get_transient( self::KICK_LOCK ) ) {
			return;
		}
		set_transient( self::KICK_LOCK, time(), 10 );

		/**
		 * Filter whether the on-demand loopback worker is used (false = rely on the scheduled poll only).
		 *
		 * @param bool $enabled Enabled.
		 */
		if ( ! apply_filters( 'streamsera_np_worker_loopback', true ) ) {
			return;
		}
		wp_remote_post(
			admin_url( 'admin-ajax.php' ),
			array(
				'timeout'   => 0.01,
				'blocking'  => false,
				// Loopback request to this site's own admin-ajax.php, made the same way as core's spawn_cron().
				'sslverify' => apply_filters( 'https_local_ssl_verify', false ), // phpcs:ignore WordPress.NamingConventions.PrefixAllGlobals.NonPrefixedHooknameFound -- WordPress core filter.
				'body'      => array(
					'action' => self::AJAX_ACTION,
					'token'  => self::token(),
				),
			)
		);
	}

	/**
	 * Loopback endpoint: run the worker.
	 */
	public function ajax_worker() {
		// No user session on a loopback request, so no nonce or capability: the request must carry a
		// secret HMAC'd with this site's salts (self::token()). The worker only refreshes now playing
		// for published stations already queued, so it exposes and changes nothing else.
		$token = isset( $_POST['token'] ) ? sanitize_text_field( wp_unslash( $_POST['token'] ) ) : ''; // phpcs:ignore WordPress.Security.NonceVerification.Missing -- authenticated by the HMAC token checked below.
		if ( ! $token || ! ( hash_equals( self::token(), $token ) || hash_equals( self::token( time() - DAY_IN_SECONDS ), $token ) ) ) {
			wp_die( '', '', array( 'response' => 403 ) );
		}
		// Let the caller go; it never waits for us anyway.
		if ( function_exists( 'fastcgi_finish_request' ) ) {
			fastcgi_finish_request();
		}
		self::drain();
		wp_die();
	}

	/**
	 * Shared secret for the loopback request (rotates daily).
	 *
	 * @param int|null $time Timestamp of the day to use (default: now).
	 * @return string
	 */
	private static function token( $time = null ) {
		return wp_hash( 'streamsera_np_worker|' . gmdate( 'Y-m-d', null === $time ? time() : (int) $time ) );
	}
}