From 27fb820ae9081fb56957cf75e79eccd8a99edd52 Mon Sep 17 00:00:00 2001
From: Jake Vanderwerf <get@jakevanderwerf.ca>
Date: Sat, 14 Feb 2026 19:14:48 +0000
Subject: [PATCH] =minor css changes, queue merge logic restructured, and double checking changes actually changed in cRUD.js before sending to server
---
inc/managers/queue/Storage.php | 79 ++++++++++++++++++++++++++++++++++++---
1 files changed, 72 insertions(+), 7 deletions(-)
diff --git a/inc/managers/queue/Storage.php b/inc/managers/queue/Storage.php
index 99cd8c3..3c580df 100644
--- a/inc/managers/queue/Storage.php
+++ b/inc/managers/queue/Storage.php
@@ -30,7 +30,7 @@
);
}
- public function fetchRunnable(int $limit = 10): array
+ public function fetchRunnable(int $offset = 0): array
{
$now = current_time('mysql');
@@ -43,18 +43,48 @@
ORDER BY
FIELD(priority, 'high', 'normal', 'low'),
scheduled_at
- LIMIT %d
+ LIMIT 10 OFFSET %d
FOR UPDATE SKIP LOCKED
- ", $now, $limit)
+ ", $now, $offset)
);
- return array_map([$this, 'rowToOperation'], $rows ?: []);
+ $total = count($rows);
+ foreach ($rows as $row) {
+ $dependencies = json_decode($row->dependencies ?? '[]', true) ?: [];
+ if (empty($dependencies)) {
+ return [$this->rowToOperation($row)];
+ }
+ $totalDep = count($dependencies);
+ $completed = [];
+ $notCompleted = [];
+ foreach ($dependencies as $dep) {
+ $dependency = $this->find($dep);
+ if ($dependency) {
+ if ($dependency->state === 'completed') {
+ $completed[] = $dep;
+ } else {
+ $notCompleted[] = $dep;
+ }
+ }
+ }
+ if (count($completed) === $totalDep) {
+ return [$this->rowToOperation($row)];
+ }
+ }
+ //If we didn't find any operations from that 10 that are ready to go, or their dependencies aren't met, try the next 10
+ if ($total === 10) {
+ return $this->fetchRunnable($offset + 10);
+ }
+
+ //If, for whatever reason, nothing still is found, there likely are none
+ return [];
}
+
public function markProcessing(string $id): bool
{
$now = current_time('mysql');
@@ -94,6 +124,7 @@
'metadata' => json_encode($op->metadata),
'result' => $op->result ? json_encode($op->result) : null,
'dependencies' => json_encode($op->dependencies),
+ 'merged_into' => $op->merged_into,
'user_dismissed' => $op->userDismissed ? 1 : 0,
'updated_at' => current_time('mysql'),
];
@@ -138,6 +169,9 @@
return false;
}
+
+ $this->invalidateUser($op->userId);
+
return true;
}
@@ -159,7 +193,7 @@
'state' => 'completed',
'outcome' => $op->outcome?? 'success',
'processed_items'=> $op->processedItems ?? 0,
- 'failed_items' => $op->failedItems ?? null,
+ 'failed_items' => $op->failedItems ? json_encode($op->failedItems) : null,
'result' => isset($op->result) ? wp_json_encode($op->result) : null,
'completed_at' => $op->completedAt ?? current_time('mysql'),
'updated_at' => current_time('mysql'),
@@ -172,7 +206,6 @@
];
$updated = $wpdb->update($table, $data, $where);
- $this->invalidateQueueCache();
if ($updated === 0) {
return true;
@@ -182,6 +215,8 @@
error_log('[Storage::saveFinal] DB error: ' . $wpdb->last_error);
return false;
}
+ $this->invalidateQueueCache();
+ $this->invalidateUser($op->userId);
return true;
}
@@ -209,6 +244,7 @@
'result' => null,
'dependencies' => json_encode($op->dependencies),
'user_dismissed' => 0,
+ 'merged_into' => null,
'created_at' => current_time('mysql'),
'updated_at' => current_time('mysql'),
]);
@@ -239,6 +275,8 @@
$type, $userId
));
+ $this->invalidateUser($userId);
+
return $row ? $this->rowToOperation($row) : null;
}
@@ -342,6 +380,7 @@
$op->startedAt = $row->started_at;
$op->completedAt = $row->completed_at;
$op->result = $row->result ? json_decode($row->result, true) : null;
+ $op->merged_into = $row->merged_into;
$op->userDismissed = (bool) $row->user_dismissed;
return $op;
@@ -442,7 +481,7 @@
private function invalidateUser(int $userId): void
{
- $this->cache->forget($userId);
+ Cache::for($userId.'_queue')->flush();
}
public function getLastError(): string
{
@@ -501,4 +540,30 @@
return (int) $affected;
});
}
+
+ /**
+ * @throws \Throwable
+ */
+ public function replaceDependency(string $fromId, string $toId): int
+ {
+ return $this->withTransaction(function () use ($fromId, $toId) {
+
+ // Only affect pending/scheduled operations
+ $affected = $this->wpdb->query($this->wpdb->prepare("
+ UPDATE {$this->table}
+ SET dependencies = REPLACE(dependencies, %s, %s),
+ updated_at = %s
+ WHERE state IN ('pending', 'scheduled')
+ AND dependencies LIKE %s
+ ",
+ '"' . $fromId . '"',
+ '"' . $toId . '"',
+ current_time('mysql'),
+ '%"' . $fromId . '"%'
+ ));
+
+ return (int) $affected;
+ });
+ }
+
}
--
Gitblit v1.10.0