<?php
/**
 * Queue worker.
 *
 * @package R2Clouder
 */

namespace R2Clouder\Queue;

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

use R2Clouder\Options;
use WP_Error;

/**
 * Drains the queue, time-boxed and memory-aware.
 *
 * Handlers are registered per job type and receive a Job. A handler returns
 * true on success or a `WP_Error` on failure; anything thrown is caught and
 * treated as a failure so one bad attachment cannot abort the whole run.
 *
 * There is deliberately no `register_shutdown_function` anywhere in this
 * plugin. Network calls made from a shutdown handler cannot be retried, cannot
 * be logged reliably, and run even when the request has already failed.
 */
final class Worker {

	/**
	 * Transient key used as a cross-request run lock.
	 */
	const LOCK_KEY = 'r2clouder_worker_lock';

	/**
	 * Nonce action for the admin-ajax tick.
	 */
	const AJAX_NONCE = 'r2clouder_queue_tick';

	/**
	 * Ajax action name for the admin-ajax tick.
	 */
	const AJAX_ACTION = 'r2clouder_queue_tick';

	/**
	 * Fraction of the memory limit at which the run stops early.
	 */
	const MEMORY_CEILING = 0.85;

	/**
	 * Registered handlers, keyed by job type.
	 *
	 * @var array<string,callable>
	 */
	private static $handlers = array();

	/**
	 * Registers the handler for a job type, replacing any previous one.
	 *
	 * @param string   $job_type One of the Job::TYPE_* constants.
	 * @param callable $handler  Receives a Job, returns true or a WP_Error.
	 * @return void
	 */
	public static function register_handler( $job_type, $handler ): void {
		if ( ! in_array( $job_type, Job::types(), true ) || ! is_callable( $handler ) ) {
			return;
		}

		self::$handlers[ $job_type ] = $handler;
	}

	/**
	 * Whether a handler exists for a job type.
	 *
	 * @param string $job_type Job type.
	 * @return bool
	 */
	public static function has_handler( $job_type ): bool {
		return isset( self::$handlers[ $job_type ] );
	}

	/**
	 * Runs the queue until it is empty or the run budget is spent.
	 *
	 * @param array<string,mixed> $args {
	 *     Optional. Run options.
	 *
	 *     @type int  $time_limit  Seconds the run may take. Defaults to the setting.
	 *     @type int  $batch_size  Jobs claimed per iteration. Defaults to the setting.
	 *     @type int  $max_jobs    Hard cap on jobs processed. Default 0 for no cap.
	 *     @type bool $use_lock    Take the cross-request lock. Default true.
	 * }
	 * @return array<string,mixed> Run statistics, or a `locked` marker.
	 */
	public static function run( array $args = array() ): array {
		$args = array_merge(
			array(
				'time_limit' => 0,
				'batch_size' => 0,
				'max_jobs'   => 0,
				'use_lock'   => true,
			),
			$args
		);

		$time_limit = (int) $args['time_limit'] > 0 ? (int) $args['time_limit'] : (int) Options::get( 'queue_time_limit', 20 );
		$time_limit = (int) apply_filters( 'r2clouder_queue_time_limit', $time_limit, $args );
		$time_limit = min( 300, max( 5, $time_limit ) );
		$batch_size = (int) $args['batch_size'] > 0 ? (int) $args['batch_size'] : (int) Options::get( 'queue_batch_size', 20 );
		$batch_size = (int) apply_filters( 'r2clouder_queue_batch_size', $batch_size, $args );
		$batch_size = min( 200, max( 1, $batch_size ) );
		$max_jobs   = max( 0, (int) $args['max_jobs'] );
		$use_lock   = ! empty( $args['use_lock'] );

		$stats = array(
			'locked'             => false,
			'claimed'            => 0,
			'done'               => 0,
			'retrying'           => 0,
			'failed'             => 0,
			'released'           => 0,
			'persistence_errors' => 0,
			'stopped'            => 'empty',
			'elapsed'            => 0.0,
		);

		if ( $use_lock && ! self::acquire_lock( $time_limit ) ) {
			$stats['locked']  = true;
			$stats['stopped'] = 'locked';

			return $stats;
		}

		$started = microtime( true );

		try {
			$stats['released'] = Queue::reap_stuck();

			while ( true ) {
				if ( microtime( true ) - $started >= $time_limit ) {
					$stats['stopped'] = 'time';
					break;
				}

				if ( self::memory_exhausted() ) {
					$stats['stopped'] = 'memory';
					break;
				}

				if ( $max_jobs > 0 && $stats['claimed'] >= $max_jobs ) {
					$stats['stopped'] = 'max_jobs';
					break;
				}

				$limit = $batch_size;

				if ( $max_jobs > 0 ) {
					$limit = min( $limit, $max_jobs - $stats['claimed'] );
				}

				$jobs = Queue::claim( $limit );

				if ( empty( $jobs ) ) {
					$stats['stopped'] = 'empty';
					break;
				}

				foreach ( $jobs as $index => $job ) {
					++$stats['claimed'];
					do_action( 'r2clouder_job_claimed', $job );

					// Anything still claimed when the budget runs out goes back to
					// the pool immediately rather than waiting to be reaped.
					if ( $index > 0 && ( microtime( true ) - $started >= $time_limit || self::memory_exhausted() ) ) {
						Queue::release( $job->id );
						--$stats['claimed'];
						continue;
					}

					$result = self::handle( $job );

					if ( true === $result ) {
						if ( Queue::complete( $job->id ) ) {
							do_action( 'r2clouder_job_completed', $job );
							++$stats['done'];
							continue;
						}

						$result = new WP_Error( 'r2clouder_completion_failed', __( 'The job finished, but its completion could not be saved.', 'devsyed-media-offloader' ) );
					}

					if ( is_wp_error( $result ) && self::is_permanent( $result ) ) {
						if ( ! Queue::fail_permanently( $job, $result ) ) {
							++$stats['persistence_errors'];
							continue;
						}
						do_action( 'r2clouder_job_failed', $job, $result, true );
						++$stats['failed'];
						continue;
					}

					$status = Queue::fail( $job, is_wp_error( $result ) ? $result : __( 'The job handler reported an unspecified failure.', 'devsyed-media-offloader' ) );

					if ( Job::STATUS_FAILED === $status ) {
						do_action( 'r2clouder_job_failed', $job, $result, true );
						++$stats['failed'];
					} elseif ( Job::STATUS_PENDING === $status ) {
						do_action( 'r2clouder_job_retry_scheduled', $job, $result, Queue::backoff_delay( $job->attempts ) );
						++$stats['retrying'];
					} else {
						// Leave the claim for bounded stuck-job recovery; its
						// failure/retry outcome has not reached the database.
						++$stats['persistence_errors'];
					}
				}
			}
		} finally {
			if ( $use_lock ) {
				self::release_lock();
			}
		}

		$stats['elapsed'] = round( microtime( true ) - $started, 3 );

		/**
		 * Fires after a queue run finishes.
		 *
		 * @param array<string,mixed> $stats Run statistics.
		 */
		do_action( 'r2clouder_queue_run_complete', $stats );

		return $stats;
	}

	/**
	 * Dispatches one job to its handler.
	 *
	 * @param Job $job Job to run.
	 * @return true|WP_Error
	 */
	public static function handle( Job $job ) {
		/**
		 * Filters the handler map before dispatch.
		 *
		 * @param array<string,callable> $handlers Handlers keyed by job type.
		 */
		$handlers = apply_filters( 'r2clouder_job_handlers', self::$handlers );

		if ( ! is_array( $handlers ) || ! isset( $handlers[ $job->job_type ] ) || ! is_callable( $handlers[ $job->job_type ] ) ) {
			return new WP_Error(
				'r2clouder_no_handler',
				/* translators: %s: job type. */
				sprintf( __( 'No handler is registered for the "%s" job type.', 'devsyed-media-offloader' ), $job->job_type )
			);
		}

		try {
			$result = call_user_func( $handlers[ $job->job_type ], $job );
		} catch ( \Throwable $e ) {
			return new WP_Error( 'r2clouder_handler_exception', get_class( $e ) . ': ' . $e->getMessage() );
		}

		if ( true === $result ) {
			return true;
		}

		if ( is_wp_error( $result ) ) {
			return $result;
		}

		return new WP_Error( 'r2clouder_handler_failed', __( 'The job handler did not report success.', 'devsyed-media-offloader' ) );
	}

	/**
	 * Cron callback.
	 *
	 * @return void
	 */
	public static function run_scheduled(): void {
		if ( 'ajax' === Options::get( 'sync_mode' ) ) {
			return;
		}

		// Jobs already in the table still run when offloading is switched off:
		// deletes, pull-backs and purges have to finish the work they started.
		if ( ! Queue::has_due_jobs() ) {
			Queue::reap_stuck();

			return;
		}

		self::run();
		Queue::purge_completed( 7 );
	}

	/**
	 * Admin-ajax callback used to tick the queue while a bulk page is open.
	 *
	 * The nonce is verified before any other request value is read, and the
	 * nonce itself is unslashed and sanitised first.
	 *
	 * @return void
	 */
	public static function ajax_tick(): void {
		$nonce = isset( $_REQUEST['_wpnonce'] ) ? sanitize_text_field( wp_unslash( $_REQUEST['_wpnonce'] ) ) : '';

		if ( ! wp_verify_nonce( $nonce, self::AJAX_NONCE ) ) {
			wp_send_json_error( array( 'message' => __( 'The security check failed. Reload the page and try again.', 'devsyed-media-offloader' ) ), 403 );
		}

		$capability = apply_filters( 'r2clouder_admin_capability', 'manage_options', 'queue' );
		$capability = is_string( $capability ) && '' !== sanitize_key( $capability ) ? sanitize_key( $capability ) : 'manage_options';

		if ( ! current_user_can( $capability ) ) {
			wp_send_json_error( array( 'message' => __( 'You are not allowed to run the queue.', 'devsyed-media-offloader' ) ), 403 );
		}

		if ( 'cron' === Options::get( 'sync_mode' ) ) {
			wp_send_json_success(
				array(
					'counts'  => Queue::counts(),
					'skipped' => true,
				)
			);
		}

		$time_limit = isset( $_REQUEST['time_limit'] ) ? absint( $_REQUEST['time_limit'] ) : 0;

		$stats           = self::run( array( 'time_limit' => $time_limit ) );
		$stats['counts'] = Queue::counts();

		wp_send_json_success( $stats );
	}

	/**
	 * Whether a failure can never succeed on a retry.
	 *
	 * A deleted attachment, a disallowed MIME type or a missing handler will look
	 * exactly the same five attempts later, so the job is parked at once instead
	 * of occupying the queue for an hour of backoff.
	 *
	 * @param WP_Error $error Failure returned by a handler.
	 * @return bool
	 */
	private static function is_permanent( WP_Error $error ): bool {
		if ( 'r2clouder_no_handler' === $error->get_error_code() ) {
			return true;
		}

		$data = $error->get_error_data();

		return is_array( $data ) && ( ! empty( $data['permanent'] ) || ( isset( $data['retryable'] ) && false === $data['retryable'] ) );
	}

	/**
	 * Takes the run lock.
	 *
	 * Uses `wp_cache_add` when a real persistent object cache is available -
	 * that is atomic - and falls back to a transient otherwise.
	 *
	 * @param int $time_limit Run budget, used to size the lock's lifetime.
	 * @return bool False when another worker holds the lock.
	 */
	private static function acquire_lock( $time_limit ): bool {
		$ttl = (int) $time_limit + MINUTE_IN_SECONDS;

		if ( wp_using_ext_object_cache() ) {
			return (bool) wp_cache_add( self::LOCK_KEY, time(), 'r2clouder', $ttl );
		}

		if ( false !== get_transient( self::LOCK_KEY ) ) {
			return false;
		}

		set_transient( self::LOCK_KEY, time(), $ttl );

		return true;
	}

	/**
	 * Releases the run lock.
	 *
	 * @return void
	 */
	private static function release_lock(): void {
		if ( wp_using_ext_object_cache() ) {
			wp_cache_delete( self::LOCK_KEY, 'r2clouder' );

			return;
		}

		delete_transient( self::LOCK_KEY );
	}

	/**
	 * Whether the process is close enough to the memory limit to stop.
	 *
	 * @return bool
	 */
	private static function memory_exhausted(): bool {
		$limit = self::memory_limit_bytes();

		if ( $limit <= 0 ) {
			return false;
		}

		return memory_get_usage( true ) >= ( $limit * self::MEMORY_CEILING );
	}

	/**
	 * Resolves the PHP memory limit in bytes.
	 *
	 * @return int Zero when the limit is unlimited or unreadable.
	 */
	private static function memory_limit_bytes(): int {
		$raw = ini_get( 'memory_limit' );

		if ( ! is_string( $raw ) || '' === $raw || '-1' === trim( $raw ) ) {
			return 0;
		}

		return (int) wp_convert_hr_to_bytes( $raw );
	}
}