<?php
/**
 * Queue dispatch triggers.
 *
 * @package Rahkar\WP
 */

declare( strict_types = 1 );

namespace Rahkar\WP\Queue;

use Rahkar\WP\Support\Context;
use Rahkar\WP\Support\Logger;

defined( 'ABSPATH' ) || exit;

/**
 * Decides when the queue actually gets drained.
 *
 * Three independent triggers, in the order they are preferred:
 *
 * 1. **Post-response drain.** On `shutdown` the response is closed with
 *    `fastcgi_finish_request()` and jobs are processed with the visitor already
 *    gone. This is the primary path and needs no server configuration at all —
 *    which matters because the sites this plugin targets have WP-Cron off and
 *    frequently no shell access either.
 * 2. **Real server cron.** A `curl` line hitting {@see WorkerEndpoint}. The only
 *    trigger that still fires when a site gets no traffic, so scheduled work
 *    (retention purges, delayed reminders) stays on time.
 * 3. **Non-blocking loopback.** Fired when urgent work is queued and no worker
 *    has run recently.
 *
 * The triggers are deliberately redundant. Each one alone is enough; together
 * they mean a misconfigured host degrades latency rather than silently dropping
 * a customer's order SMS.
 */
final class Dispatcher {

	public const LAST_RUN_OPTION = 'rahkar_wp_queue_last_run';

	/**
	 * Jobs processed in one drain.
	 */
	private const BATCH_SIZE = 20;

	/**
	 * Wall-clock ceiling for a post-response drain, in seconds.
	 *
	 * Kept short because the php-fpm worker is still occupied even though the
	 * visitor has their bytes — holding it for a minute would starve the pool
	 * under load.
	 */
	private const INLINE_BUDGET = 10.0;

	/**
	 * Wall-clock ceiling for a dedicated worker request, in seconds.
	 */
	private const WORKER_BUDGET = 50.0;

	/**
	 * Seconds before a job nobody handled is offered again.
	 *
	 * Short enough that a job claimed by the wrong kind of request — a front-end
	 * drain picking up the background-only log purge — reaches the right one
	 * soon; long enough that a job with no handler anywhere costs one UPDATE per
	 * quarter hour rather than one per page view.
	 */
	private const UNHANDLED_DELAY = 900;

	/**
	 * Minimum gap between loopback pokes, in seconds.
	 */
	private const LOOPBACK_COOLDOWN = 60;

	/**
	 * The one dispatcher.
	 *
	 * @var self|null
	 */
	private static ?self $instance = null;

	/**
	 * Guards against a second drain in the same request.
	 *
	 * Two paths can reach {@see self::drainAfterResponse()} — the core bootstrap's
	 * shutdown hook and {@see Queue::push()} on a request that queued something —
	 * and the response can only be closed once.
	 *
	 * @var bool
	 */
	private bool $drained = false;

	/**
	 * Shared instance.
	 */
	public static function instance(): self {
		if ( null === self::$instance ) {
			self::$instance = new self();
		}

		return self::$instance;
	}

	/**
	 * Note that work was queued during this request.
	 *
	 * Called by {@see Queue::push()}. The drain itself is already hooked to
	 * `shutdown` by the core bootstrap, so this only has to decide whether the
	 * work is urgent enough to also poke the loopback trigger.
	 *
	 * @param bool $urgent Whether the work should not wait for the next visitor.
	 */
	public function schedule( bool $urgent = false ): void {
		if ( $urgent && ! Context::is( Context::ANY_BACKGROUND ) ) {
			add_action( 'shutdown', [ $this, 'pokeLoopback' ], 5 );
		}
	}

	/**
	 * Close the response, then work the queue.
	 *
	 * @internal Hooked callback.
	 */
	public function drainAfterResponse(): void {
		// The response can only be closed once, and two paths reach this method.
		if ( $this->drained ) {
			return;
		}

		$this->drained = true;

		if ( ! $this->shouldDrainInline() ) {
			return;
		}

		$this->closeResponse();

		$this->run( self::BATCH_SIZE, self::INLINE_BUDGET );
	}

	/**
	 * Drain the queue. Used by the worker endpoint and WP-CLI.
	 *
	 * @param int   $limit  Maximum jobs to process.
	 * @param float $budget Wall-clock ceiling in seconds.
	 *
	 * @return array{processed:int, failed:int}
	 */
	public function run( int $limit = self::BATCH_SIZE, float $budget = self::WORKER_BUDGET ): array {
		$started   = microtime( true );
		$processed = 0;
		$failed    = 0;

		Queue::recoverStalled();

		update_option( self::LAST_RUN_OPTION, time(), false );

		foreach ( Queue::claim( $limit ) as $row ) {
			if ( ( microtime( true ) - $started ) > $budget ) {
				// Hand the rest back rather than risking a timeout mid-job.
				Queue::release( (int) $row['id'], (int) $row['attempts'], 'Worker budget exhausted before start.' );

				continue;
			}

			// Nothing is listening in this request. Hand it back for a request that
			// does register the handler, instead of firing a hook into the void
			// and deleting the job as though it had run.
			if ( ! has_action( (string) $row['hook'] ) ) {
				Queue::defer( $row, self::UNHANDLED_DELAY );

				continue;
			}

			if ( $this->runJob( $row ) ) {
				++$processed;
			} else {
				++$failed;
			}
		}

		/**
		 * Fires after every queue run, worker or inline.
		 *
		 * For self-arming periodic jobs. A job that re-queues itself at the end of
		 * each run has one way to get started — something has to queue the first
		 * one — and until 0.28.0 that something was an admin opening one of the
		 * plugin's screens. A store that switched a module on and never opened
		 * those screens never got the job at all. Re-arming here, with the
		 * idempotent `Queue::pushUnique()`, costs one indexed lookup per run and
		 * only in requests that are already doing queue work.
		 *
		 * @since 0.28.0
		 */
		do_action( 'rahkar_wp_queue_ran' );

		// Correct the watermark now that the table has changed, so an emptied
		// queue stops waking the dispatcher on subsequent requests.
		Queue::refreshWatermark();

		return [
			'processed' => $processed,
			'failed'    => $failed,
		];
	}

	/**
	 * Execute one job.
	 *
	 * @param array<string,mixed> $row Job row.
	 *
	 * @return bool True when the job completed.
	 */
	private function runJob( array $row ): bool {
		$id       = (int) $row['id'];
		$hook     = (string) $row['hook'];
		$attempts = (int) $row['attempts'] + 1;

		$payload = json_decode( (string) $row['payload'], true );
		$payload = is_array( $payload ) ? $payload : [];

		try {
			/**
			 * Fires when a queued job runs.
			 *
			 * The hook name is whatever was passed to {@see Queue::push()}; handlers
			 * are registered by the module that queued the work.
			 *
			 * @param array<string,mixed> $payload Job arguments.
			 */
			// phpcs:ignore WordPress.NamingConventions.PrefixAllGlobals.DynamicHooknameFound -- The hook is whatever the caller queued; jobs of add-ons carry their own prefixes.
			do_action( $hook, $payload );

			Queue::complete( $id );

			return true;
		} catch ( \Throwable $e ) {
			Queue::release( $id, $attempts, $e->getMessage() );

			Logger::error(
				'queue',
				'Job threw an exception.',
				[
					'id'    => $id,
					'hook'  => $hook,
					'error' => $e->getMessage(),
				]
			);

			return false;
		}
	}

	/**
	 * Is an inline drain appropriate for this request?
	 */
	private function shouldDrainInline(): bool {
		// A cron or CLI context has its own drain path.
		if ( Context::is( Context::ANY_BACKGROUND ) ) {
			return false;
		}

		// Answered from the autoloaded watermark — no query. On a site with an
		// empty queue this is where every request stops, which is what keeps the
		// drain hook free to register unconditionally.
		if ( ! Queue::maybeHasDue() ) {
			return false;
		}

		/**
		 * Filters whether the queue may drain after this response.
		 *
		 * Lets a site opt out on specific requests — for instance to keep a
		 * latency-sensitive REST endpoint's worker free.
		 *
		 * @param bool $allowed Whether to drain.
		 */
		return (bool) apply_filters( 'rahkar_wp_allow_inline_drain', true );
	}

	/**
	 * Send the response and detach from the visitor.
	 *
	 * `fastcgi_finish_request()` is the clean path and exists on php-fpm, which is
	 * effectively universal on the hosts this targets. The fallback flushes the
	 * buffers and relies on `Connection: close`, which is less reliable but still
	 * better than making the visitor wait on an SMS API.
	 */
	private function closeResponse(): void {
		ignore_user_abort( true );

		if ( function_exists( 'fastcgi_finish_request' ) ) {
			fastcgi_finish_request();

			return;
		}

		if ( function_exists( 'litespeed_finish_request' ) ) {
			litespeed_finish_request();

			return;
		}

		if ( ! headers_sent() ) {
			header( 'Connection: close' );
			header( 'Content-Encoding: none' );
		}

		while ( ob_get_level() > 0 ) {
			ob_end_flush();
		}

		flush();
	}

	/**
	 * Poke the worker endpoint without waiting for it.
	 *
	 * Third trigger. Used when work is queued that should not wait for the next
	 * visitor — and rate-limited, because a burst of orders must not turn into a
	 * burst of loopback requests.
	 */
	public function pokeLoopback(): void {
		$last = (int) get_option( self::LAST_RUN_OPTION, 0 );

		if ( ( time() - $last ) < self::LOOPBACK_COOLDOWN ) {
			return;
		}

		// A request to this same site. Certificate verification follows the filter
		// WordPress core uses for its own loopbacks (`spawn_cron()`, the site
		// health checks), rather than being switched off here: local and staging
		// sites often run self-signed certificates, and a host that verifies its
		// own certificate can say so in one place for every loopback.
		wp_remote_post(
			WorkerEndpoint::url(),
			[
				'timeout'   => 0.01,
				'blocking'  => false,
				// phpcs:ignore WordPress.NamingConventions.PrefixAllGlobals.NonPrefixedHooknameFound -- WordPress core's own filter, used as core uses it.
				'sslverify' => (bool) apply_filters( 'https_local_ssl_verify', false ),
			]
		);
	}

	/**
	 * Seconds since the queue last ran, or null when it never has.
	 */
	public static function secondsSinceLastRun(): ?int {
		$last = (int) get_option( self::LAST_RUN_OPTION, 0 );

		return $last > 0 ? time() - $last : null;
	}
}