Test/queue runner refacto

inc/Engine/Admin/RocketInsights/Queue/QueueRunner.php

@@ -0,0 +1,120 @@
+<?php
+declare( strict_types=1 );
+
+namespace WP_Rocket\Engine\Admin\RocketInsights\Queue;
+
+use WP_Rocket\Engine\Common\Queue\AbstractQueueRunner;
+use WP_Rocket\Engine\Common\Queue\Cleaner;
+use ActionScheduler_Store;
+use ActionScheduler_FatalErrorMonitor;
+use ActionScheduler_AsyncRequest_QueueRunner;
+
+/**
+ * Rocket Insights Queue Runner
+ *
+ * Manages Action Scheduler jobs for Rocket Insights workflow
+ * Processes jobs every 30 seconds instead of the default 60 seconds
+ *
+ * @since 3.20
+ */
+class QueueRunner extends AbstractQueueRunner {
+
+	/**
+	 * Cron hook name.
+	 */
+	const WP_CRON_HOOK = 'action_scheduler_run_queue_rocket_insights';
+
+	/**
+	 * Cron schedule interval.
+	 */
+	const WP_CRON_SCHEDULE = 'every_thirty_seconds';
+
+	/**
+	 * Current runner instance.
+	 *
+	 * @var QueueRunner Instance.
+	 */
+	private static $runner = null;
+
+	/**
+	 * Get singleton instance.
+	 *
+	 * @since 3.20
+	 *
+	 * @return QueueRunner Instance.
+	 */
+	public static function instance() {
+		if ( null === self::$runner ) {
+			self::$runner = new QueueRunner();
+		}
+		return self::$runner;
+	}
+
+	/**
+	 * Add the cron schedule for Rocket Insights (every 30 seconds).
+	 *
+	 * @since 3.20
+	 *
+	 * @param array $schedules Array of current schedules.
+	 *
+	 * @return array
+	 */
+	public function add_wp_cron_schedule( $schedules ) {
+		if ( isset( $schedules['every_thirty_seconds'] ) ) {
+			return $schedules;
+		}
+
+		$schedules['every_thirty_seconds'] = [
+			'interval' => 30, // in seconds.
+			'display'  => __( 'Every thirty seconds', 'rocket' ),
+		];
+
+		return $schedules;
+	}
+
+	/**
+	 * Get the cron hook name.
+	 *
+	 * @since 3.20
+	 *
+	 * @return string
+	 */
+	protected function get_wp_cron_hook(): string {
+		return self::WP_CRON_HOOK;
+	}
+
+	/**
+	 * Get the cron schedule interval.
+	 *
+	 * @since 3.20
+	 *
+	 * @return string
+	 */
+	protected function get_wp_cron_schedule(): string {
+		return self::WP_CRON_SCHEDULE;
+	}
+
+	/**
+	 * Get the queue group name.
+	 *
+	 * @since 3.20
+	 *
+	 * @return string
+	 */
+	protected function get_group(): string {
+		return 'rocket-insights';
+	}
+
+	/**
+	 * Process a batch of jobs.
+	 *
+	 * @param int    $size    The maximum number of actions to process.
+	 * @param string $context Optional identifer for the context in which this action is being processed, e.g. 'WP CLI'.
+	 *
+	 * @return int The number of actions processed.
+	 */
+	protected function do_batch( $size = 100, $context = '' ) {
+		$processed = parent::do_batch( $size, $context );
+		return $processed;
+	}
+}

inc/Engine/Admin/RocketInsights/ServiceProvider.php

@@ -15,6 +15,7 @@
 	Jobs\Manager as RIManager,
 	Managers\Plan,
 	Queue\Queue as RIQueue,
+	Queue\QueueRunner as RIQueueRunner,
 	URLLimit\Subscriber as URLLimitSubscriber,
 	Settings\Controller as SettingsController,
 	Settings\Subscriber as SettingsSubscriber,
@@ -180,9 +181,7 @@ public function register(): void {
 					'ri_manager',
 					'ri_plan',
 				]
-			);
-
-		// URL Limit subscriber.
+			);        // URL Limit subscriber.
 		$this->getContainer()->addShared( 'ri_url_limit_subscriber', URLLimitSubscriber::class )
 			->addArguments(
 				[

inc/Engine/Admin/RocketInsights/Subscriber.php

@@ -9,6 +9,7 @@
 	Managers\Plan,
 	Jobs\Manager,
 	Queue\Queue,
+	Queue\QueueRunner as RIQueueRunner,
 };
 use WP_Rocket\Admin\Options_Data;
 use WP_Rocket\Event_Management\Subscriber_Interface;
@@ -85,7 +86,6 @@ class Subscriber implements Subscriber_Interface, LoggerAwareInterface {
 	 * @var Plan
 	 */
 	private $plan;
-
 	/**
 	 * Constructor.
 	 *
@@ -153,6 +153,7 @@ public static function get_subscribed_events(): array {
 			'admin_init'                            => [
 				[ 'flush_license_cache', 8 ],
 				[ 'check_upgrade' ],
+				[ 'initialize_ri_queue_runner', 10 ],
 				[ 'schedule_jobs', 11 ],
 			],
 			'admin_post_rocket_rocket_insights_add_homepage' => 'add_homepage_from_widget',
@@ -425,6 +426,21 @@ public function check_upgrade() {
 		$this->plan->check_upgrade();
 	}
 
+	/**
+	 * Initialize the queue runner for Rocket Insights.
+	 *
+	 * @since 3.20
+	 *
+	 * @return void
+	 */
+	public function initialize_ri_queue_runner() {
+		if ( ! $this->context->is_allowed() ) {
+			return;
+		}
+
+		RIQueueRunner::instance()->init();
+	}
+
 	/**
 	 * Remove current plan with plugin deactivation.
 	 *

inc/Engine/Common/JobManager/JobProcessor.php

@@ -20,13 +20,6 @@ class JobProcessor implements LoggerAwareInterface {
 	 */
 	private $factories;
 
-	/**
-	 * Queue instance.
-	 *
-	 * @var Queue
-	 */
-	private $queue;
-
 	/**
 	 * Retry Strategy Factory
 	 *
@@ -45,18 +38,15 @@ class JobProcessor implements LoggerAwareInterface {
 	 * Instantiate the class.
 	 *
 	 * @param array           $factories Array of factories.
-	 * @param Queue           $queue Queue instance.
 	 * @param StrategyFactory $strategy_factory Strategy Factory.
 	 * @param WPRClock        $clock Clock object instance.
 	 */
 	public function __construct(
 		array $factories,
-		Queue $queue,
 		StrategyFactory $strategy_factory,
 		WPRClock $clock
 	) {
 		$this->factories        = $factories;
-		$this->queue            = $queue;
 		$this->strategy_factory = $strategy_factory;
 		$this->wpr_clock        = $clock;
 	}
@@ -134,7 +124,9 @@ public function process_pending_jobs() {
 				$optimization_type = $this->get_optimization_type( $row );
 				// Change status to in-progress.
 				$this->make_status_inprogress( $row->url, $row->is_mobile, $optimization_type );
-				$this->queue->add_job_status_check_async( $row->url, $row->is_mobile, $optimization_type );
+				// Use the appropriate queue for the optimization type to ensure correct group assignment.
+				$queue = Queue::get_for_optimization_type( $optimization_type );
+				$queue->add_job_status_check_async( $row->url, $row->is_mobile, $optimization_type );
 			}
 		}
 

inc/Engine/Common/JobManager/Queue/Queue.php

@@ -21,6 +21,15 @@ class Queue extends AbstractASQueue {
 	 */
 	protected $group = 'rocket-rucss';
 
+	/**
+	 * Constructor.
+	 *
+	 * @param string $group Queue group name.
+	 */
+	public function __construct( string $group = 'rocket-rucss' ) {
+		$this->group = $group;
+	}
+
 	/**
 	 * Pending jobs cron hook.
 	 *
@@ -76,4 +85,39 @@ public function add_job_status_check_async( string $url, bool $is_mobile, string
 			]
 		);
 	}
+
+	/**
+	 * Create a queue instance for RUCSS optimization.
+	 *
+	 * @return Queue
+	 */
+	public static function create_for_rucss(): Queue {
+		return new Queue( 'rocket-rucss' );
+	}
+
+	/**
+	 * Create a queue instance for Rocket Insights optimization.
+	 *
+	 * @return Queue
+	 */
+	public static function create_for_rocket_insights(): Queue {
+		return new Queue( 'rocket-insights' );
+	}
+
+	/**
+	 * Get the appropriate queue instance for an optimization type.
+	 *
+	 * @param string $optimization_type The optimization type.
+	 *
+	 * @return Queue
+	 */
+	public static function get_for_optimization_type( string $optimization_type ): Queue {
+		switch ( $optimization_type ) {
+			case 'rocket_insights':
+				return self::create_for_rocket_insights();
+			case 'rucss':
+			default:
+				return self::create_for_rucss();
+		}
+	}
 }

inc/Engine/Common/JobManager/ServiceProvider.php

@@ -62,7 +62,6 @@ public function register(): void {
 			->addArguments(
 				[
 					$factories,
-					'queue',
 					'retry_strategy_factory',
 					'wpr_clock',
 				]