<?php
/**
 * Engine: scan queue - a crash-safe, time-boxed, instantiable queue runner.
 *
 * Queue-based bulk processing per the cross-plugin standard: the queue is an
 * option of opaque items; batches are processed by browser-driven AJAX calls
 * while a scan page is open, by a SELF-DISPATCHED LOOPBACK CHAIN when running
 * unattended (SPEC-queue-loopback-chain.md - each batch fires a non-blocking
 * single-use-token self-request that runs the next batch immediately, so the
 * queue drains at wall-clock speed regardless of the site's cron cadence), and
 * by a WP-Cron continuation event as the WATCHDOG that restarts a dead chain.
 * Each batch is time-boxed, schedules its continuation BEFORE the loop, and
 * always processes at least one item.
 *
 * Generalized from SSAA's static singleton SSAA_Scan_Queue into a runner
 * configured per instance (option key, cron hook, lock, budgets, retries,
 * processor, on_done). It is PHASE-AGNOSTIC: it runs one queue, processes each
 * item through the injected processor callable, and fires on_done when the
 * queue drains. It does not know whether a consumer chains several queues -
 * that sequencing lives entirely in the consumer (e.g. LIHA_Scan). This is the
 * one deliberate departure from the static-only class convention, confined to
 * the engine queue.
 *
 * Items are opaque: scalars (post IDs) or arrays (page descriptors, URL
 * batches). The runner never inspects an item beyond deriving a stable key for
 * retry bookkeeping; producing and consuming items is the processor's job.
 *
 * CANONICAL ENGINE SOURCE - edit here in _shared/shared-render-engine/src/, re-run sync.sh.
 *
 * @package plugupp-render-engine
 */

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

class SSAA_Engine_Queue {

	/** Seconds a stale lock survives (covers one full web-cron budget). */
	const LOCK_TTL = 90;

	/** Chain-hop token lifetime in seconds. Tokens are single-use regardless
	 * (the render door's token model - see SPEC-queue-loopback-chain.md). */
	const CHAIN_TOKEN_TTL = 60;

	/** Default politeness pause before a chain hop processes (seconds). Keeps
	 * the chain from turning the old ~3% cron duty cycle into a sustained
	 * 100% self-render load on small shared hosts. Filterable per queue via
	 * '{prefix}_chain_hop_delay' (clamped 0-30). */
	const CHAIN_HOP_DELAY = 2;

	/** Default time budget per web-cron tick (seconds). */
	const WEB_BUDGET = 60;

	/** Default time budget per AJAX batch (seconds). */
	const AJAX_BUDGET = 12;

	/** Default per-item retry budget (extra attempts beyond the first). */
	const MAX_RETRIES = 2;

	/** @var array Instance config. */
	private $cfg;

	/**
	 * @param array $config {
	 *     @type string   $name        Queue name (for logs / filters).
	 *     @type string   $option_key  Option storing the queue array. REQUIRED.
	 *     @type string   $cron_hook   Cron continuation hook. REQUIRED.
	 *     @type string   $lock_key    Concurrency lock transient key. REQUIRED.
	 *     @type int      $web_budget  Seconds per web-cron tick.
	 *     @type int      $ajax_budget Seconds per AJAX batch.
	 *     @type int      $max_retries Extra attempts for transient failures.
	 *     @type callable $processor   fn( mixed $item ): array outcome. REQUIRED.
	 *     @type callable $on_done     fn( array $stats ): void  - queue drained.
	 *     @type callable $on_progress fn( array $state ): void  - after a non-final batch.
	 *     @type callable $on_cancel   fn(): void                - scan cancelled.
	 * }
	 */
	public function __construct( array $config ) {
		$this->cfg = $config;
	}

	/**
	 * Register the cron continuation hook. Called once at boot per instance.
	 */
	public function init() {
		add_action( $this->cfg['cron_hook'], array( $this, 'cron_tick' ) );
	}

	// -------------------------------------------------------------------------
	// Config accessors
	// -------------------------------------------------------------------------

	private function name() {
		return isset( $this->cfg['name'] ) ? (string) $this->cfg['name'] : (string) $this->cfg['option_key'];
	}

	public function web_budget() {
		return max( 1, (int) ( isset( $this->cfg['web_budget'] ) ? $this->cfg['web_budget'] : self::WEB_BUDGET ) );
	}

	public function ajax_budget() {
		return max( 1, (int) ( isset( $this->cfg['ajax_budget'] ) ? $this->cfg['ajax_budget'] : self::AJAX_BUDGET ) );
	}

	/**
	 * Retry budget per item (>= 0). Filterable via '{prefix}_scan_max_retries'.
	 *
	 * @return int
	 */
	private function max_retries() {
		$default = isset( $this->cfg['max_retries'] ) ? (int) $this->cfg['max_retries'] : self::MAX_RETRIES;
		return max( 0, (int) apply_filters( SSAA_Engine::key( 'scan_max_retries' ), $default, $this->name() ) );
	}

	// -------------------------------------------------------------------------
	// Lifecycle
	// -------------------------------------------------------------------------

	/**
	 * Build and store a fresh queue, schedule the cron continuation. Replaces
	 * any existing queue on this instance.
	 *
	 * @param array  $items  Opaque items to process.
	 * @param string $origin Informational tag passed through to on_done.
	 * @param bool   $full   Whether this run covers the full discoverable set
	 *                       (on_done can then prune untouched rows).
	 * @return array Queue state.
	 */
	public function start_with( array $items, $origin = 'manual', $full = false ) {
		$items = array_values( $items );

		update_option(
			$this->cfg['option_key'],
			array(
				'items'      => $items,
				'total'      => count( $items ),
				'origin'     => sanitize_key( $origin ),
				'full'       => (bool) $full,
				'started_at' => time(),
			),
			false
		);

		$this->schedule_continuation();

		SSAA_Logger::info( 'queue_started', array(
			'queue' => $this->name(),
			'total' => count( $items ),
		) );

		// Chain kick-off: fire the first hop immediately instead of waiting a
		// dead minute for the cron continuation. Harmless when a browser AJAX
		// chain also drives this queue - the lock arbitrates.
		if ( count( $items ) > 0 ) {
			$this->dispatch_chain();
		}

		return $this->state();
	}

	/**
	 * Cancel a running scan: queue, lock and continuation removed; results from
	 * already-processed items are kept. Fires the on_cancel callback so the
	 * consumer can refresh its own summary.
	 */
	public function cancel() {
		delete_option( $this->cfg['option_key'] );
		delete_transient( $this->cfg['lock_key'] );
		wp_clear_scheduled_hook( $this->cfg['cron_hook'] );

		$this->fire( 'on_cancel' );

		SSAA_Logger::info( 'queue_cancelled', array( 'queue' => $this->name() ) );
	}

	/**
	 * Current queue state.
	 *
	 * @return array { running, total, remaining, done }
	 */
	public function state() {
		$queue = get_option( $this->cfg['option_key'], array() );

		if ( empty( $queue ) || ! isset( $queue['items'] ) ) {
			return array(
				'running'   => false,
				'total'     => 0,
				'remaining' => 0,
				'done'      => 0,
			);
		}

		$total     = (int) $queue['total'];
		$remaining = count( (array) $queue['items'] );

		$state = array(
			'running'   => true,
			'total'     => $total,
			'remaining' => $remaining,
			'done'      => max( 0, $total - $remaining ),
		);

		// Deferred-queue surface (0.4.0): lets the admin UI say "paused -
		// resuming shortly" and lets the AJAX driver slow its polling instead
		// of busy-looping against a paused queue.
		if ( ! empty( $queue['deferred_until'] ) && time() < (int) $queue['deferred_until'] ) {
			$state['deferred']     = true;
			$state['deferred_for'] = (int) $queue['deferred_until'] - time();
		}

		return $state;
	}

	// -------------------------------------------------------------------------
	// Processing
	// -------------------------------------------------------------------------

	/**
	 * WP-Cron entry point - the chain WATCHDOG. CLI cron gets an unbounded
	 * budget; web loopback cron is capped (standard pattern - see _standards
	 * DECISIONS.md). If this tick did real time-boxed work with items left,
	 * the chain died (or never started) - hand off to a fresh hop.
	 */
	public function cron_tick() {
		$budget = ( 'cli' === php_sapi_name() ) ? PHP_INT_MAX : $this->web_budget();
		$state  = $this->process_batch( (int) apply_filters( SSAA_Engine::key( 'queue_time_budget' ), $budget, $this->name() ) );

		if ( $this->should_dispatch( $state ) ) {
			$this->dispatch_chain();
		}
	}

	/**
	 * Chain-hop entry point (SPEC-queue-loopback-chain.md): politeness pause,
	 * one web-budget batch, then self-dispatch the next hop while time-boxed
	 * work remains. Always the WEB budget, never the CLI unbounded budget -
	 * each hop is a web request, and fresh hops replace long execution.
	 */
	public function chain_hop() {
		// The pause runs BEFORE the lock is taken, in a dedicated non-blocking
		// request - it costs nothing and keeps the duty cycle polite.
		$delay = (int) apply_filters( SSAA_Engine::key( 'chain_hop_delay' ), self::CHAIN_HOP_DELAY, $this->name() );
		$delay = max( 0, min( 30, $delay ) );
		if ( $delay > 0 ) {
			sleep( $delay );
		}

		$state = $this->process_batch( (int) apply_filters( SSAA_Engine::key( 'queue_time_budget' ), $this->web_budget(), $this->name() ) );

		if ( $this->should_dispatch( $state ) ) {
			$this->dispatch_chain();
		}
	}

	/**
	 * Process queue items until the queue empties or the budget elapses. Always
	 * processes at least one item. Safe against concurrent callers (AJAX chain +
	 * cron tick + chain hop) via a lock transient.
	 *
	 * @param int $budget Seconds available.
	 * @return array Queue state plus 'processed' (count for this call) and
	 *               'stopped' (why the loop ended: 'idle' | 'locked' |
	 *               'drained' | 'deadline' | 'retry_guard') - the chain
	 *               dispatch decision reads 'stopped' (see should_dispatch()).
	 */
	public function process_batch( $budget ) {
		$state = $this->state();

		if ( ! $state['running'] ) {
			$state['processed'] = 0;
			$state['stopped']   = 'idle';
			return $state;
		}

		// Watchdog re-arm BEFORE the lock check. When WP-Cron fires the
		// continuation event, core removes it from the cron array before the
		// hook runs - so a tick that then finds the lock held would consume
		// the event with nothing rescheduling it (the lock holder's own
		// schedule_continuation() was a no-op back when the event still
		// existed). On page-load-cron hosting that strands a running queue
		// whose last batch ends on deferred retries (found live on 20i,
		// 2026-07-14). Re-arming on EVERY entry while the queue runs makes
		// the watchdog unconditional; the completion path still clears it.
		$this->schedule_continuation();

		// Concurrency guard: if another runner holds the lock, report state
		// without processing - the AJAX chain just polls until the lock frees.
		if ( false !== get_transient( $this->cfg['lock_key'] ) ) {
			$state['processed'] = 0;
			$state['stopped']   = 'locked';
			return $state;
		}

		$queue_meta = get_option( $this->cfg['option_key'], array() );

		// Deferred-queue guard (0.4.0): the whole queue is paused (site
		// maintenance / retry-later signal). Report without processing; the
		// cron watchdog (re-armed above) re-enters after the window. No lock
		// taken - nothing runs.
		if ( ! empty( $queue_meta['deferred_until'] ) && time() < (int) $queue_meta['deferred_until'] ) {
			$state['processed'] = 0;
			$state['stopped']   = 'deferred';
			return $state;
		}

		set_transient( $this->cfg['lock_key'], 1, self::LOCK_TTL );

		$deadline  = microtime( true ) + max( 1, (int) $budget );
		$processed = 0;
		$stopped   = 'drained';
		$attempted = array(); // Item keys already tried in THIS batch (retry guard).

		while ( true ) {
			$queue = get_option( $this->cfg['option_key'], array() );
			if ( empty( $queue['items'] ) ) {
				break;
			}

			// If the front item is one we already tried this batch, the only
			// work left is deferred retries - stop so they fall to a later tick
			// and the server gets recovery time before we re-hit it. The chain
			// deliberately does NOT re-dispatch off this stop (see
			// should_dispatch()) - recovery pacing stays with the cron watchdog.
			if ( in_array( self::item_key( $queue['items'][0] ), $attempted, true ) ) {
				$stopped = 'retry_guard';
				break;
			}

			$item           = array_shift( $queue['items'] );
			$queue['items'] = array_values( $queue['items'] );
			// Persist the pop before processing - crash safety over double-scan
			// safety (a re-process of one item is harmless).
			update_option( $this->cfg['option_key'], $queue, false );

			$key      = self::item_key( $item );
			$result   = (array) $this->run_processor( $item );
			$attempted[] = $key;
			$processed++;

			// Deferrable outcome (0.4.0): the SITE told us to come back later
			// (maintenance mode, 503/429). Pause the WHOLE queue: the item
			// goes back to the FRONT with no retry consumed, a queue-level
			// deferred_until stops all processing until the window passes, and
			// the cron watchdog resumes. Total deferral is capped (default 15
			// minutes, filter '{prefix}_scan_max_defer'); past the cap the
			// outcome degrades to an ordinary retryable failure so a scan can
			// never hang forever.
			if ( ! empty( $result['deferrable'] ) ) {
				$defer_started = isset( $queue['defer_started'] ) ? (int) $queue['defer_started'] : 0;
				$max_defer     = max( 60, (int) apply_filters( SSAA_Engine::key( 'scan_max_defer' ), 15 * MINUTE_IN_SECONDS, $this->name() ) );

				if ( 0 === $defer_started || ( time() - $defer_started ) < $max_defer ) {
					$defer_for = isset( $result['defer_seconds'] ) ? min( 600, max( 30, (int) $result['defer_seconds'] ) ) : 90;

					array_unshift( $queue['items'], $item );
					$queue['items']         = array_values( $queue['items'] );
					$queue['deferred_until'] = time() + $defer_for;
					if ( 0 === $defer_started ) {
						$queue['defer_started'] = time();
					}
					update_option( $this->cfg['option_key'], $queue, false );

					SSAA_Logger::info( 'queue_deferred', array(
						'queue'     => $this->name(),
						'defer_for' => $defer_for,
						'code'      => isset( $result['error_code'] ) ? (string) $result['error_code'] : '',
					) );

					$stopped = 'deferred';
					break;
				}

				// Defer budget exhausted: degrade to the retryable path below.
				SSAA_Logger::info( 'queue_defer_exhausted', array( 'queue' => $this->name() ) );
				$result['retryable'] = true;
			} elseif ( isset( $queue['deferred_until'] ) || isset( $queue['defer_started'] ) ) {
				// First healthy outcome after a defer window: clear the defer
				// state so a later, separate maintenance window gets a fresh
				// deferral budget.
				unset( $queue['deferred_until'], $queue['defer_started'] );
				update_option( $this->cfg['option_key'], $queue, false );
			}

			// Permanent failure (0.4.0): count it so on_done stats can report
			// scan health honestly ('full' only ever meant coverage, not
			// success - the 2026-07-28 incident lesson).
			if ( isset( $result['status'] ) && 'error' === (string) $result['status'] && empty( $result['retryable'] ) ) {
				$queue['failures'] = isset( $queue['failures'] ) ? (int) $queue['failures'] + 1 : 1;
				update_option( $this->cfg['option_key'], $queue, false );
			}

			// Transient failure: defer to the BACK of the queue until this
			// item's retry budget is spent.
			if ( ! empty( $result['retryable'] ) ) {
				$retries  = isset( $queue['retries'] ) ? (array) $queue['retries'] : array();
				$so_far   = isset( $retries[ $key ] ) ? (int) $retries[ $key ] : 0;
				$err_code = isset( $result['error_code'] ) ? (string) $result['error_code'] : '';

				if ( $so_far < $this->max_retries() ) {
					$retries[ $key ]  = $so_far + 1;
					$queue['retries'] = $retries;
					$queue['items'][] = $item;
					update_option( $this->cfg['option_key'], $queue, false );

					SSAA_Logger::info( 'queue_item_retry_deferred', array(
						'queue'   => $this->name(),
						'attempt' => $so_far + 1,
						'max'     => $this->max_retries(),
						'code'    => $err_code,
					) );
				} else {
					SSAA_Logger::info( 'queue_item_retries_exhausted', array(
						'queue' => $this->name(),
						'code'  => $err_code,
					) );
					// The item is dropped - that is a permanent failure for
					// this scan; count it toward the health stats (0.4.0).
					$queue['failures'] = isset( $queue['failures'] ) ? (int) $queue['failures'] + 1 : 1;
					update_option( $this->cfg['option_key'], $queue, false );
				}
			}

			if ( empty( $queue['items'] ) ) {
				break;
			}
			if ( microtime( true ) >= $deadline ) {
				$stopped = 'deadline';
				break;
			}
		}

		$state = $this->state();

		if ( $state['running'] && 0 === $state['remaining'] ) {
			// Queue drained: clean up, then hand off to the consumer. Re-read
			// the option first - failure counts accumulated during THIS batch
			// live in the stored queue, not in the stale pre-loop $queue_meta.
			$final_meta = get_option( $this->cfg['option_key'], array() );
			if ( ! is_array( $final_meta ) || empty( $final_meta ) ) {
				$final_meta = $queue_meta;
			}
			delete_option( $this->cfg['option_key'] );
			wp_clear_scheduled_hook( $this->cfg['cron_hook'] );

			$stats = array(
				'queue'      => $this->name(),
				'total'      => (int) $state['total'],
				'origin'     => isset( $final_meta['origin'] ) ? (string) $final_meta['origin'] : 'manual',
				// 'full' means "this run covered the discoverable set" -
				// COVERAGE, not success. Health lives in 'failures' (0.4.0).
				'full'       => ! empty( $final_meta['full'] ),
				'failures'   => isset( $final_meta['failures'] ) ? (int) $final_meta['failures'] : 0,
				'deferred'   => ! empty( $final_meta['defer_started'] ),
				'started_at' => isset( $final_meta['started_at'] ) ? (int) $final_meta['started_at'] : 0,
			);

			SSAA_Logger::info( 'queue_completed', array(
				'queue'    => $stats['queue'],
				'total'    => $stats['total'],
				'origin'   => $stats['origin'],
				'failures' => $stats['failures'],
			) );

			$this->fire( 'on_done', $stats );

			$state = $this->state();
		} else {
			// Let the consumer keep its dashboard summary roughly current.
			$this->fire( 'on_progress', $state );
		}

		delete_transient( $this->cfg['lock_key'] );

		$state['processed'] = $processed;
		$state['stopped']   = $stopped;
		return $state;
	}

	// -------------------------------------------------------------------------
	// Loopback chain - hop endpoint + tokens (SPEC-queue-loopback-chain.md)
	// -------------------------------------------------------------------------

	/**
	 * Validate and run chain-hop requests. Requests WITHOUT the chain query var
	 * are untouched. Requests with it must carry a valid single-use token that
	 * was minted for the named queue, or the request dies with 403 - like the
	 * render door, the endpoint never leaks anything on bad input.
	 *
	 * Registered on parse_request (priority 0) beside the door, from
	 * SSAA_Engine::boot(). The hop response body is irrelevant - the
	 * dispatcher fires it non-blocking and never reads it.
	 */
	public static function guard_chain() {
		$qv_chain = SSAA_Engine::key( 'chain' );
		$qv_token = SSAA_Engine::key( 'chain_token' );

		// phpcs:ignore WordPress.Security.NonceVerification.Recommended -- the single-use token IS the credential; this is not a form action.
		if ( ! isset( $_GET[ $qv_chain ] ) ) {
			return;
		}

		// phpcs:ignore WordPress.Security.NonceVerification.Recommended -- see above.
		$name = sanitize_key( wp_unslash( $_GET[ $qv_chain ] ) );
		// phpcs:ignore WordPress.Security.NonceVerification.Recommended -- see above.
		$token = isset( $_GET[ $qv_token ] ) ? sanitize_text_field( wp_unslash( $_GET[ $qv_token ] ) ) : '';

		// Consume FIRST (single use even on mismatch), then resolve the queue -
		// a token must never survive a failed hop.
		$valid = self::consume_chain_token( $token, $name );
		$queue = SSAA_Engine::queue( $name );

		if ( ! $valid || ! $queue ) {
			SSAA_Logger::info( 'chain_hop_rejected', array( 'queue' => $name ) );
			wp_die(
				esc_html__( 'Invalid chain request.', 'search-social-appearance-audit' ),
				'',
				array( 'response' => 403 )
			);
		}

		$queue->chain_hop();
		exit;
	}

	/**
	 * Whether a finished batch should self-dispatch the next hop. Only a
	 * DEADLINE stop with real work done qualifies:
	 *
	 * - 'locked': another runner is processing; IT dispatches at its own end.
	 *   Re-dispatching here would busy-wait the held lock at hop-delay cadence.
	 * - 'retry_guard': the only work left is deferred retries; leave them to
	 *   the cron watchdog so the server gets the designed recovery time.
	 * - 'drained' / 'idle': nothing left to do.
	 *
	 * @param array $state process_batch() return value.
	 * @return bool
	 */
	private function should_dispatch( array $state ) {
		return ! empty( $state['running'] )
			&& (int) $state['remaining'] > 0
			&& (int) $state['processed'] > 0
			&& isset( $state['stopped'] ) && 'deadline' === $state['stopped'];
	}

	/**
	 * Fire the next chain hop: a non-blocking loopback self-request carrying a
	 * fresh single-use token. Failure is silent BY DESIGN - the cron watchdog
	 * (schedule_continuation) restores exactly the pre-chain cadence, so a
	 * host that blocks rapid self-requests degrades to today's behaviour,
	 * never worse.
	 */
	private function dispatch_chain() {
		/**
		 * Kill switch: disable the loopback chain for this queue (reverts to
		 * the cron-only continuation model).
		 *
		 * @param bool   $enabled Default true.
		 * @param string $name    Queue name.
		 */
		if ( ! apply_filters( SSAA_Engine::key( 'chain_enabled' ), true, $this->name() ) ) {
			return;
		}

		$url = add_query_arg(
			array(
				SSAA_Engine::key( 'chain' )       => rawurlencode( $this->name() ),
				SSAA_Engine::key( 'chain_token' ) => $this->mint_chain_token(),
			),
			home_url( '/' )
		);

		/**
		 * Dispatch timeout for the non-blocking hop request. Some transports
		 * need >= 1s for a non-blocking request to actually leave the box;
		 * raise via this filter if hops never fire on a host.
		 *
		 * @param float  $timeout Seconds. Default 0.01.
		 * @param string $name    Queue name.
		 */
		$timeout = (float) apply_filters( SSAA_Engine::key( 'chain_dispatch_timeout' ), 0.01, $this->name() );

		wp_remote_post( $url, array(
			'blocking'  => false,
			'timeout'   => max( 0.01, $timeout ),
			'sslverify' => false, // Self-request; mirrors the renderer's loopback conventions.
		) );

		SSAA_Logger::debug( 'queue_chain_dispatched', array(
			'queue'     => $this->name(),
			'remaining' => (int) $this->state()['remaining'],
		) );
	}

	/**
	 * Issue a single-use chain-hop token for this queue.
	 *
	 * @return string
	 */
	public function mint_chain_token() {
		$token = bin2hex( random_bytes( 16 ) );
		set_transient( self::chain_token_transient( $token ), $this->name(), self::CHAIN_TOKEN_TTL );
		return $token;
	}

	/**
	 * Validate and consume (delete) a chain token. Valid only when it exists,
	 * has not expired, and was minted for this exact queue.
	 *
	 * @param string $token Presented token.
	 * @param string $name  Requested queue name.
	 * @return bool
	 */
	private static function consume_chain_token( $token, $name ) {
		if ( '' === $token || ! preg_match( '/^[a-f0-9]{32}$/', $token ) ) {
			return false;
		}

		$stored = get_transient( self::chain_token_transient( $token ) );
		delete_transient( self::chain_token_transient( $token ) ); // Single use, even on mismatch.

		return false !== $stored && '' !== (string) $stored && (string) $stored === (string) $name;
	}

	/**
	 * Transient key for a chain token, namespaced by consumer prefix (the
	 * door's token model; '_ct_' beside the door's '_rt_').
	 *
	 * @param string $token Token.
	 * @return string
	 */
	private static function chain_token_transient( $token ) {
		return SSAA_Engine::prefix() . '_ct_' . $token;
	}

	// -------------------------------------------------------------------------
	// Internals
	// -------------------------------------------------------------------------

	/**
	 * Invoke the processor for one item. Returns its outcome array, or a safe
	 * "not retryable error" if no valid processor is configured.
	 *
	 * @param mixed $item Queue item.
	 * @return array Outcome.
	 */
	private function run_processor( $item ) {
		if ( empty( $this->cfg['processor'] ) || ! is_callable( $this->cfg['processor'] ) ) {
			SSAA_Logger::error( 'queue_no_processor', array( 'queue' => $this->name() ) );
			return array( 'status' => 'error', 'retryable' => false );
		}
		$outcome = call_user_func( $this->cfg['processor'], $item );
		return is_array( $outcome ) ? $outcome : array( 'status' => 'ok', 'retryable' => false );
	}

	/**
	 * Fire an optional config callback if present and callable.
	 *
	 * @param string $name Callback config key.
	 * @param mixed  $arg  Single argument to pass.
	 */
	private function fire( $name, $arg = null ) {
		if ( ! empty( $this->cfg[ $name ] ) && is_callable( $this->cfg[ $name ] ) ) {
			call_user_func( $this->cfg[ $name ], $arg );
		}
	}

	/**
	 * Schedule the cron continuation if one is not already pending.
	 */
	private function schedule_continuation() {
		if ( ! wp_next_scheduled( $this->cfg['cron_hook'] ) ) {
			wp_schedule_single_event( time() + MINUTE_IN_SECONDS, $this->cfg['cron_hook'] );
		}
	}

	/**
	 * Stable key for an item, for retry bookkeeping. Scalars key as
	 * themselves; arrays key by a hash of their content.
	 *
	 * @param mixed $item Queue item.
	 * @return string
	 */
	private static function item_key( $item ) {
		if ( is_scalar( $item ) ) {
			return (string) $item;
		}
		return 'h:' . md5( (string) wp_json_encode( $item ) );
	}
}
