=Cleaning up the Integrations setup a bit, started the Helcim integration w...

Jake Vanderwerf

2026-09-18 a6413da67a4579866f0fd0323ddb23a7b6b9654a

 inc/managers/queue/Queue.php

@@ -4,22 +4,33 @@
   exit;
}

use JVBase\managers\CustomTable;
use Exception;
use JVBase\inc\managers\ErrorHelper;
use JVBase\Instance;
use JVBase\managers\AdminPages;
use JVBase\managers\Errors;
use JVBase\rest\routes\QueueRoutes;
use Throwable;
use WP_Error;
use WP_REST_Request;
use WP_REST_Response;

class Queue
{
   use Instance, ErrorHelper;
   private Storage $storage;
   private Processor $processor;
   private TypeRegistry $registry;
   public Executor $executor;
   private Locker $locker;

   public function __construct()
   private QueueRoutes $routes;

   private function __construct()
   {
      $this->defineTables();
      //Load routes
      $this->routes = new QueueRoutes();

      $this->storage = new Storage();
      $this->registry = new TypeRegistry();
      $this->locker = new Locker();
@@ -29,6 +40,7 @@

      add_action('jvb_process_queue', [$this, 'checkQueue']);
      add_action('jvb_queue_maintenance', [$this, 'maintenance']);
      add_action('jvb_daily_snapshot', [$this->storage, 'snapshotDaily']);

      if (!wp_next_scheduled('jvb_process_queue')) {
         wp_schedule_event(time(), 'every-minute', 'jvb_process_queue');
@@ -36,89 +48,16 @@
      if (!wp_next_scheduled('jvb_queue_maintenance')) {
         wp_schedule_event(time(), 'hourly', 'jvb_queue_maintenance');
      }
      if (!wp_next_scheduled('jvb_daily_snapshot')) {
         // Schedule for next 3am
         $next3am = strtotime('tomorrow 3am', current_time('timestamp'));
         wp_schedule_event($next3am, 'daily', 'jvb_daily_snapshot');
      }

      jvb_register_do_once('queue_admin_action_registered', [$this, 'registerAdminAction']);
      add_filter(BASE.'admin_action_filter', [$this, 'adminActionFilter'], 10, 3);
   }

   public static function defineTables():void
   {
      $queue = CustomTable::for('_operation_queue');
      $queue->setColumns([
         'id'        => 'VARCHAR(64) NOT NULL',
         'type'         => 'VARCHAR(50) NOT NULL',
         'user_id'      => $queue->getUserIDType().' NOT NULL',

         'request_data' => 'JSON NOT NULL CHECK (JSON_VALID(request_data))',

         'total_items'  => 'INT(11) NOT NULL DEFAULT 1',
         'processed_items' => 'INT(11) DEFAULT 0',
         'failed_items' => 'JSON',

         'priority'     => 'ENUM(\'high\',\'normal\',\'low\') DEFAULT \'normal\'',
         'state'        => 'ENUM(\'pending\', \'scheduled\', \'processing\', \'completed\') DEFAULT \'pending\'',
         'outcome'      => 'ENUM(\'pending\', \'success\',\'partial\',\'merged\',\'failed\',\'failed_permanent\') DEFAULT \'pending\'',

         'retries'      => 'INT(11) DEFAULT 0',
         'last_error_hash'=> 'CHAR(32) DEFAULT NULL',
         'error_message'   => 'TEXT',

         'scheduled_at' => 'DATETIME DEFAULT NULL',
         'started_at'   => 'DATETIME DEFAULT CURRENT_TIMESTAMP',
         'completed_at' => 'DATETIME DEFAULT NULL',

         'metadata'     => 'JSON DEFAULT NULL',
         'result'    => 'JSON',
         'dependencies' => 'JSON',
         'merged_into'  => 'VARCHAR(64) DEFAULT NULL',

         'user_dismissed'=> 'tinyint(1) DEFAULT 0',
         'created_at'   => 'DATETIME DEFAULT CURRENT_TIMESTAMP',
         'updated_at'   => 'DATETIME DEFAULT CURRENT_TIMESTAMP',
      ]);

      $queue->setKeys([
         ['key' => 'PRIMARY', 'value' => 'id'],
         '`idx_run_queue` (`state`, `priority`, `scheduled_at`)',
         '`idx_user_ops` (`user_id`, `state`)',
         '`idx_user_type_pending` (`user_id`, `type`, `state`)',
         '`idx_completed_at` (`completed_at`)',
         '`idx_processing_stuck` (`state`, `started_at`)'
      ]);

      $queue->defineTable();

      $stats = CustomTable::for('stats__operation_queue');
      $stats->setColumns([
         'id'  => 'BIGINT unsigned AUTO_INCREMENT',
         'date'   => 'DATE NOT NULL',
         'type'   => 'VARCHAR(50) NOT NULL',

         'total_operations'      => 'INT NOT NULL DEFAULT 0',
         'successful_operations' => 'INT NOT NULL DEFAULT 0',
         'partial_operations' => 'INT NOT NULL DEFAULT 0',
         'failed_operations'     => 'INT NOT NULL DEFAULT 0',
         'failed_permanent_operations'=> 'INT NOT NULL DEFAULT 0',
         'total_items_processed' => 'INT NOT NULL DEFAULT 0',

         'average_duration'      => 'FLOAT DEFAULT NULL',
         'max_duration'       => 'INT DEFAULT NULL',
         'peak_queue_size'    => 'INT NOT NULL DEFAULT 0',
         'peak_memory_usage'     => 'INT DEFAULT NULL',
         'peak_cpu_usage'     => 'FLOAT DEFAULT NULL',

         'created_at'         => 'TIMESTAMP DEFAULT CURRENT_TIMESTAMP',
      ]);

      $stats->setKeys([
         ['key' => 'PRIMARY', 'value' => 'id'],
         ['key' => 'UNIQUE', 'value' => '(`date`, `type`)'],
         '`date_idx` (`date`)',
         '`type_idx` (`type`)'
      ]);
      $stats->defineTable();
   }

   /**
    * Access type registry for registering operation configs
    */
@@ -139,13 +78,15 @@
    * @param int $userId User ID
    * @param array $data Request data
    * @param array $options {
    *     @type string $priority     'low', 'normal', 'high'
    *     @type int $delay           Seconds to delay processing
    *     @type string $scheduled    Specific datetime to process
    *     @type array|string $depends_on  Operation IDs this depends on
    *     @type string|array $chunk_key   Key(s) to chunk (overrides registry)
    *     @type int $chunk_size      Chunk size (overrides registry)
    * @type string $priority 'low', 'normal', 'high'
    * @type int $delay Seconds to delay processing
    * @type string $scheduled Specific datetime to process
    * @type array|string $depends_on Operation IDs this depends on
    * @type string|array $chunk_key Key(s) to chunk (overrides registry)
    * @type int $chunk_size Chunk size (overrides registry)
    * }
    * @throws Throwable
    * @throws Throwable
    */
   public function add(string $type, int $userId, array $data, array $options = []): array|WP_Error
   {
@@ -168,8 +109,9 @@
            'updated_existing' => false,
         ];

      } catch (\Exception $e) {
         JVB()->error()->log('queue', $e->getMessage(), $data, 'high');
      } catch (Exception $e) {
         $this->logError($e->getMessage(), 'add', $data);

         return new WP_Error('queue_failed', $e->getMessage());
      }
   }
@@ -177,7 +119,7 @@
   /**
    * Try to merge incoming operation into an existing pending/scheduled one.
    * Returns result array if merged, null if not.
    * @throws \Throwable
    * @throws Throwable
    */
   private function tryMerge(Operation $incoming): ?array
   {
@@ -246,7 +188,7 @@
   public function maintenance(): void
   {
      $this->locker->withLock(function () {
         $this->storage->resetStuckOperations(30);
         $this->storage->resetStuckOperations();
      });
      $this->runQueueOnShutdown();
   }
@@ -269,7 +211,7 @@
   /**
    * Get a specific value from an operation - backwards compatibility
    */
   public function getOperationValue(string $id, string $column, bool $decodeJson = true): mixed
   public function getOperationValue(string $id, string $column, bool $decodeJson = true): static|string|int|array|null
   {
      $op = $this->get($id);
      if (!$op) {
@@ -416,7 +358,7 @@
      // Use provided operation_id or generate one
      $op->id = !empty($options['operation_id'])
         ? $options['operation_id']
         : 'u' . $userId . '_' . uniqid('op_');
         : self::generateID($userId);

      $op->type = $type;
      $op->userId = $userId;
@@ -477,7 +419,7 @@

   private function runQueueOnShutdown(): void
   {
      if (!has_action('shutdown', [$this, 'processQueueOnShutdown'])) {
      if (!has_action('shutdown', [$this, 'processQueueOnShutdown']) && $this->processor->hasAdequateResources()) {
         add_action('shutdown', [$this, 'processQueueOnShutdown'], 100);
      }
   }
@@ -495,7 +437,7 @@

   public function registerAdminAction():void
   {
      $admin = JVB()->admin();
      $admin = AdminPages::getInstance();
      $admin->registerAction(
         'Restart Stuck Operations',
         'restart-stuck-operations',
@@ -509,11 +451,12 @@
         'infinity'
      );
   }

   /**
    * @param WP_REST_Response $response
    * @param WP_REST_Request $request
    * @param string $action
    *
    * @return bool|WP_REST_Response
    * @return WP_REST_Response|bool
    */
   public function adminActionFilter(WP_REST_Response $response, WP_REST_Request $request, string $action):WP_REST_Response|bool
   {
@@ -536,4 +479,9 @@
            return $response;
      }
   }

   public static function generateId(int $userID):string
   {
      return 'u' . $userID . '_' . uniqid('op_');
   }
}