From 772462eeca3002a1d52508aeba485aab2b4742ad Mon Sep 17 00:00:00 2001
From: Jake Vanderwerf <get@jakevanderwerf.ca>
Date: Tue, 03 Mar 2026 19:06:19 +0000
Subject: [PATCH] =MAJOR OVERHAUL. Likely should have made a new branch ages ago. Key changes: Registrar.php is the base for custom post types, taxonomies, and user roles. Replaces JVB_CONTENT, JVB_TAXONOMY, and JVB_USER constants, eliminates most of Features.php (except for JVB_SITE, JVB_MEMBERSHIP), and has built in sanitizing and validation via sub-classes. Also started a major overhaul of the Schema output. Created a shit ton of property traits and classes to help sanitize and ensure proper data for different schema types. Still a bunch to do, but better to be starting committing changes here on this other branch.
---
inc/managers/queue/Queue.php | 204 ++++++++++++++++++++++++++++++++++++++++++++++++--
1 files changed, 195 insertions(+), 9 deletions(-)
diff --git a/inc/managers/queue/Queue.php b/inc/managers/queue/Queue.php
index 54509fa..d93078b 100644
--- a/inc/managers/queue/Queue.php
+++ b/inc/managers/queue/Queue.php
@@ -4,23 +4,28 @@
exit;
}
+use JVBase\managers\CustomTable;
use WP_Error;
+use WP_REST_Request;
+use WP_REST_Response;
class Queue
{
private Storage $storage;
private Processor $processor;
private TypeRegistry $registry;
+ public Executor $executor;
private Locker $locker;
public function __construct()
{
+ $this->defineTables();
$this->storage = new Storage();
$this->registry = new TypeRegistry();
$this->locker = new Locker();
- $executor = new FilteredExecutor($this->storage);
- $this->processor = new Processor($this->storage, $executor, $this->registry);
+ $this->executor = new FilteredExecutor();
+ $this->processor = new Processor($this->storage, $this->executor, $this->registry);
add_action('jvb_process_queue', [$this, 'checkQueue']);
add_action('jvb_queue_maintenance', [$this, 'maintenance']);
@@ -31,6 +36,87 @@
if (!wp_next_scheduled('jvb_queue_maintenance')) {
wp_schedule_event(time(), 'hourly', 'jvb_queue_maintenance');
}
+
+ 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();
}
/**
@@ -66,6 +152,13 @@
try {
$incoming = $this->buildOperation($type, $userId, $data, $options);
+ // Attempt pre-insert merge
+ $merged = $this->tryMerge($incoming);
+ if ($merged) {
+ $this->runQueueOnShutdown();
+ return $merged;
+ }
+
$this->storage->insert($incoming);
$this->runQueueOnShutdown();
@@ -82,6 +175,59 @@
}
/**
+ * Try to merge incoming operation into an existing pending/scheduled one.
+ * Returns result array if merged, null if not.
+ * @throws \Throwable
+ */
+ private function tryMerge(Operation $incoming): ?array
+ {
+ $mergeable = $this->registry->getMergeable($incoming->type);
+ if (!$mergeable) {
+ return null;
+ }
+
+ $criteria = $mergeable->matchCriteria($incoming);
+
+ $existing = $this->storage->findMergeable($incoming->type, $incoming->userId, $criteria);
+ if (!$existing || !$mergeable->canMerge($existing, $incoming)) {
+ return null;
+ }
+
+ $this->storage->withTransaction(function () use ($incoming, $existing, $mergeable) {
+ $mergeable->merge($existing, $incoming);
+
+ $this->storage->replaceDependency(
+ $incoming->id,
+ $existing->id
+ );
+
+ // Merge dependency arrays safely
+ $existing->dependencies = array_values(array_unique(
+ array_merge($existing->dependencies, $incoming->dependencies)
+ ));
+
+ // Prevent self dependency
+ $existing->dependencies = array_diff(
+ $existing->dependencies,
+ [$existing->id]
+ );
+
+ $this->storage->save($existing);
+
+ $incoming->state = 'completed';
+ $incoming->outcome = 'merged';
+ $incoming->merged_into = $existing->id;
+ $this->storage->saveFinal($incoming);
+ });
+
+ return [
+ 'success' => true,
+ 'operation_id' => $existing->id,
+ 'updated_existing' => true,
+ ];
+ }
+
+ /**
* Alias for add() - backwards compatibility
*/
public function queueOperation(string $type, int $userId, array $data, array $options = []): array|WP_Error
@@ -100,14 +246,9 @@
public function maintenance(): void
{
$this->locker->withLock(function () {
- $this->cleanupStuck();
+ $this->storage->resetStuckOperations(30);
});
-
- }
-
- private function cleanupStuck(): void
- {
- $this->storage->resetStuckOperations(30);
+ $this->runQueueOnShutdown();
}
// === Public Getters ===
@@ -144,6 +285,7 @@
'user_id' => $op->userId,
'metadata' => $op->metadata,
'dependencies' => $op->dependencies,
+ 'merged_into' => $op->merged_into,
default => null,
};
}
@@ -350,4 +492,48 @@
$this->checkQueue();
}
+
+ public function registerAdminAction():void
+ {
+ $admin = JVB()->admin();
+ $admin->registerAction(
+ 'Restart Stuck Operations',
+ 'restart-stuck-operations',
+ 'manage_options',
+ 'arrows-clockwise'
+ );
+ $admin->registerAction(
+ 'Unlock Queue',
+ 'unlock-operation-queue',
+ 'manage_options',
+ 'infinity'
+ );
+ }
+ /**
+ * @param WP_REST_Response $response
+ * @param string $action
+ *
+ * @return bool|WP_REST_Response
+ */
+ public function adminActionFilter(WP_REST_Response $response, WP_REST_Request $request, string $action):WP_REST_Response|bool
+ {
+ switch ($action) {
+ case 'unlock-operation-queue':
+ error_log('Unlocking Queue');
+ $this->locker->unlock();
+ return new WP_REST_Response([
+ 'success' => true,
+ 'message' => 'Unlocked Queue'
+ ]);
+ case 'restart-stuck-operations':
+ error_log('Restarting stuck operations');
+ $this->maintenance();
+ return new WP_REST_Response([
+ 'success' => true,
+ 'message' => 'Restarted Stuck Operations'
+ ]);
+ default:
+ return $response;
+ }
+ }
}
--
Gitblit v1.10.0