| | |
| | | <?php |
| | | namespace JVBase\managers; |
| | | |
| | | use JVBase\managers\CacheManager; |
| | | use Exception; |
| | | use JVBase\utility\Features; |
| | | use WP_Error; |
| | | use WP_REST_Response; |
| | | use WP_REST_Request; |
| | |
| | | |
| | | ]; |
| | | |
| | | protected ?CacheManager $cache = null; |
| | | protected ?Cache $cache = null; |
| | | protected int $ttl = 300; |
| | | protected string $cacheGroup = 'queue'; |
| | | // Cache keys for different data types |
| | | private const CACHE_QUEUE_STATUS = 'status'; |
| | | private const CACHE_HAS_ITEMS = 'has_items'; |
| | | private const CACHE_OPERATION_PREFIX = 'op_'; |
| | | private const CACHE_USER_QUEUE_PREFIX = 'user_queue_'; |
| | | private const CACHE_METRICS_PREFIX = 'metrics_'; |
| | | private const CACHE_QUEUE_SIZE = 'queue_size'; |
| | | |
| | | // Prepared statement cache |
| | | protected array $preparedStatements = []; |
| | | |
| | | public function __construct() |
| | | { |
| | | global $wpdb; |
| | | $this->wpdb = $wpdb; |
| | | $this->cache = CacheManager::for('queue', DAY_IN_SECONDS); |
| | | $this->cache = Cache::for('queue', 'user',DAY_IN_SECONDS)->user(); |
| | | add_action('jvb_process_queue', [ $this, 'checkQueue' ]); |
| | | add_action('jvb_queue_maintenance', [$this, 'hourlyMaintenance']); |
| | | add_action('jvbEmailDailyMetricsReport', [$this, 'emailDailyMetricsReport']); |
| | |
| | | } |
| | | public function registerAdminAction():void |
| | | { |
| | | $admin = JVB()->admin(); |
| | | $admin = AdminPages::getInstance(); |
| | | $admin->registerAction( |
| | | 'Restart Stuck Operations', |
| | | 'restart-stuck-operations', |
| | |
| | | * |
| | | * @return bool|WP_REST_Response |
| | | */ |
| | | public function adminActionFilter(WP_REST_Response $response, WP_REST_Request $request, string $action) |
| | | public function adminActionFilter(WP_REST_Response $response, WP_REST_Request $request, string $action):WP_REST_Response|bool |
| | | { |
| | | switch ($action) { |
| | | case 'unlock-operation-queue': |
| | |
| | | $result = $this->wpdb->get_row($this->wpdb->prepare(" |
| | | SELECT |
| | | COUNT(*) as total, |
| | | SUM(CASE WHEN status IN ('pending', 'processing') THEN 1 ELSE 0 END) as active, |
| | | SUM(CASE WHEN status = 'scheduled' AND scheduled_at <= %s THEN 1 ELSE 0 END) as ready_scheduled |
| | | SUM(IF(status IN ('pending', 'processing'), 1, 0)) as active, |
| | | SUM(IF(status = 'scheduled' AND scheduled_at <= %s, 1, 0)) as ready_scheduled |
| | | FROM $table |
| | | WHERE status IN ('pending', 'processing', 'scheduled') |
| | | ", $current_time)); |
| | |
| | | // $this->pruneOldMetrics(); // Optional: clean old metrics |
| | | } |
| | | |
| | | protected function pruneOldMetrics(): void |
| | | { |
| | | $metricsTable = $this->wpdb->prefix . $this->metricsTable; |
| | | |
| | | // Keep only last 90 days of metrics |
| | | $this->wpdb->query($this->wpdb->prepare( |
| | | "DELETE FROM $metricsTable WHERE date < %s", |
| | | date('Y-m-d', strtotime('-90 days')) |
| | | )); |
| | | } |
| | | |
| | | protected function isHeavyOperation(string $type): bool |
| | | { |
| | | $heavyOperations = [ |
| | |
| | | * Check if the queue is currently locked for processing |
| | | * @return bool |
| | | */ |
| | | protected function isQueueLocked():bool |
| | | { |
| | | $lock = get_transient(BASE . 'queue_lock'); |
| | | protected function isQueueLocked(): bool |
| | | { |
| | | $lock_name = BASE . '_queue_lock'; |
| | | |
| | | if ($lock) { |
| | | // Check if the lock is stale (process might have crashed) |
| | | $lock_time = (int)$lock; |
| | | $current_time = time(); |
| | | // Check if lock exists and is recent |
| | | $result = $this->wpdb->get_var($this->wpdb->prepare( |
| | | "SELECT IS_USED_LOCK(%s)", |
| | | $lock_name |
| | | )); |
| | | |
| | | // If lock is older than 5 minutes, it's probably stale |
| | | if (($current_time - $lock_time) > 300) { |
| | | $this->unlockQueue(); // Clear stale lock |
| | | return false; |
| | | } |
| | | return true; |
| | | } |
| | | return false; |
| | | } |
| | | return $result !== null; |
| | | } |
| | | |
| | | /** |
| | | * Check if server is relatively idle (very low load) |
| | |
| | | * Acquire a lock on the queue |
| | | * @return bool |
| | | */ |
| | | protected function lockQueue():bool |
| | | { |
| | | // Try to acquire the lock |
| | | $result = set_transient(BASE . 'queue_lock', time()); |
| | | protected function lockQueue(): bool |
| | | { |
| | | $lock_name = BASE . '_queue_lock'; |
| | | |
| | | if ($result) { |
| | | return true; |
| | | } |
| | | return false; |
| | | } |
| | | // Try to acquire lock with 0 timeout (non-blocking) |
| | | $result = $this->wpdb->get_var($this->wpdb->prepare( |
| | | "SELECT GET_LOCK(%s, 0)", |
| | | $lock_name |
| | | )); |
| | | |
| | | return $result === '1'; |
| | | } |
| | | |
| | | /** |
| | | * Release the queue lock |
| | | * @return void |
| | | */ |
| | | protected function unlockQueue():void |
| | | { |
| | | delete_transient(BASE . 'queue_lock'); |
| | | } |
| | | protected function unlockQueue(): void |
| | | { |
| | | $lock_name = BASE . '_queue_lock'; |
| | | |
| | | $this->wpdb->query($this->wpdb->prepare( |
| | | "SELECT RELEASE_LOCK(%s)", |
| | | $lock_name |
| | | )); |
| | | } |
| | | |
| | | |
| | | /** |
| | |
| | | } |
| | | } |
| | | |
| | | $this->updateLastModified($user_id); |
| | | $this->invalidateQueueCache(); |
| | | $this->cache->delete(self::CACHE_USER_QUEUE_PREFIX . $user_id); |
| | | $this->invalidateUserQueue($user_id); |
| | | $this->runQueueOnShutdown(); |
| | | |
| | | return [ |
| | |
| | | ]; |
| | | |
| | | } catch (Exception $e) { |
| | | JVB()->error()->log('queue', $e->getMessage(), $data, 'high'); |
| | | Errors::getInstance()->log('queue', $e->getMessage(), $data, 'high'); |
| | | return new WP_Error('queue_failed', $e->getMessage()); |
| | | } |
| | | } |
| | | |
| | | protected function updateLastModified(int $user_id) { |
| | | CacheManager::updateTimestamp("user_{$user_id}"); |
| | | } |
| | | |
| | | protected function deepMerge(array $existing, array $new): array |
| | | { |
| | | $merged = $existing; |
| | |
| | | $this->processOperation($operation); |
| | | |
| | | // Invalidate operation cache after processing |
| | | $this->cache->delete(self::CACHE_OPERATION_PREFIX . $operation->id); |
| | | $this->cache->delete(self::CACHE_USER_QUEUE_PREFIX . $operation->user_id); |
| | | $this->cache->forget(self::CACHE_OPERATION_PREFIX . $operation->id); |
| | | $this->cache->forget(self::CACHE_USER_QUEUE_PREFIX . $operation->user_id); |
| | | } |
| | | |
| | | // Batch invalidate caches at the end |
| | |
| | | AND retries < %d |
| | | ORDER BY |
| | | FIELD(priority, 'high', 'normal', 'low'), |
| | | COALESCE(scheduled_at, created_at) ASC |
| | | COALESCE(scheduled_at, created_at) |
| | | LIMIT %d |
| | | ", $current_time, $this->max_attempts, $batch_size)); |
| | | |
| | |
| | | } |
| | | |
| | | /** |
| | | * Check if user's queue has been modified since given timestamp |
| | | * Returns true if modified, false if not (can send 304) |
| | | */ |
| | | public function isUserQueueModified(int $user_id, int $since_timestamp): bool |
| | | { |
| | | return $this->cache::lastModified("user_{$user_id}") > $since_timestamp; |
| | | } |
| | | protected function invalidateUserQueue(int $user_id): void |
| | | { |
| | | // This automatically: |
| | | // 1. Updates HTTP timestamp for user_{$user_id} |
| | | // 2. Flushes user-specific caches |
| | | // 3. Triggers connected cache invalidation |
| | | Cache::for($user_id)->flush(); |
| | | } |
| | | |
| | | /** |
| | | * Invalidate all queue-related caches |
| | | */ |
| | | protected function invalidateQueueCache(string $scope = 'all'): void |
| | |
| | | $keys = $cacheKeys[$scope] ?? $cacheKeys['all']; |
| | | |
| | | foreach ($keys as $key) { |
| | | $this->cache->delete($key); |
| | | $this->cache->forget($key); |
| | | } |
| | | |
| | | $this->cache->touch(); |
| | | |
| | | if ($scope === 'all') { |
| | | // Clear entire group for complete refresh |
| | | $this->cache->invalidate(); |
| | | delete_transient('jvb_queue_status_counts'); |
| | | } |
| | | } |
| | |
| | | ); |
| | | |
| | | // Log this as a critical error |
| | | JVB()->error()->log( |
| | | Errors::getInstance()->log( |
| | | '[OperationQueue]:restartStuckOperations', |
| | | "Operation failed after maximum retries", |
| | | [ |
| | |
| | | ); |
| | | |
| | | // Log the backoff |
| | | JVB()->error()->log( |
| | | Errors::getInstance()->log( |
| | | '[OperationQueue]:restartStuckOperations', |
| | | "Operation requeued with exponential backoff", |
| | | [ |
| | |
| | | |
| | | $message .= "Please check the error logs for more details."; |
| | | |
| | | return JVB()->email()->sendEmail($admin_email, $subject, $message); |
| | | return Emails::getInstance()->sendEmail($admin_email, $subject, $message); |
| | | } |
| | | |
| | | /** |
| | |
| | | if (!$updated) { |
| | | throw new Exception('Operation no longer available for processing'); |
| | | } |
| | | $this->updateLastModified($operation->user_id); |
| | | $this->invalidateUserQueue($operation->user_id); |
| | | |
| | | $data = json_decode($operation->request_data, true); |
| | | $progress_count = (int) $operation->progress_count; |
| | |
| | | if (!is_array($filterResult) || !isset($filterResult['success'])) { |
| | | // Log invalid filter response and quit early |
| | | $invalid_response = is_object($filterResult) ? get_class($filterResult) : gettype($filterResult); |
| | | JVB()->error()->log( |
| | | Errors::getInstance()->log( |
| | | '[OperationQueue]:processOperation', |
| | | "Invalid filter response format - expected array with 'success' key", |
| | | [ |
| | |
| | | ); |
| | | |
| | | // Now do post-completion tasks |
| | | $this->invalidateQueueCache('status'); |
| | | $this->updateLastModified($operation->user_id); |
| | | $this->updateUserQueueTimestamp($operation->user_id); |
| | | $this->invalidateUserQueue($operation->user_id); |
| | | |
| | | $this->trackOperationMetrics($operation->id); |
| | | |
| | |
| | | ['%s'] |
| | | ); |
| | | |
| | | $this->updateLastModified($operation->user_id); |
| | | $this->invalidateUserQueue($operation->user_id); |
| | | } |
| | | |
| | | } else { |
| | |
| | | ['%s'] |
| | | ); |
| | | |
| | | $this->invalidateQueueCache('status'); |
| | | $this->updateLastModified($operation->user_id); |
| | | $this->invalidateUserQueue($operation->user_id); |
| | | } else { |
| | | // Failed but more to process - continue with next chunk |
| | | $this->wpdb->update( |
| | |
| | | } |
| | | } |
| | | // Clear operation cache after any update |
| | | $this->cache->delete(self::CACHE_OPERATION_PREFIX . $operation->id); |
| | | $this->updateLastModified($operation->user_id); |
| | | $this->invalidateUserQueue($operation->user_id); |
| | | return $filterResult; |
| | | |
| | | } catch (Exception $e) { |
| | | // Handle exception... |
| | | JVB()->error()->log( |
| | | Errors::getInstance()->log( |
| | | '[OperationQueue]:processOperation', |
| | | "Exception during operation processing: " . $e->getMessage(), |
| | | [ |
| | |
| | | $batch = $this->extractBatch($data, $keys, $current_progress, $chunk_size); |
| | | |
| | | // Merge non-chunked data with chunked batch |
| | | $result = []; |
| | | foreach ($data as $k => $v) { |
| | | if (!in_array($k, $keys)) { |
| | | $result[$k] = $v; // Keep non-batch data |
| | | } |
| | | } |
| | | $result = array_filter($data, function ($k) use ($keys) { |
| | | return !in_array($k, $keys); |
| | | }, ARRAY_FILTER_USE_KEY); |
| | | |
| | | // Add the batched data |
| | | foreach ($batch as $k => $v) { |
| | |
| | | protected function updateUserQueueTimestamp(int $user_id) |
| | | { |
| | | |
| | | CacheManager::updateTimestamp("user_{$user_id}"); |
| | | Cache::touch("user_{$user_id}"); |
| | | } |
| | | |
| | | /** |
| | |
| | | |
| | | $success_rate = round(($total_success / max(1, $total_ops)) * 100, 1); |
| | | |
| | | $message = JVB()->email()->h1('Daily Queue Performance Report'); |
| | | $message = Emails::getInstance()->h1('Daily Queue Performance Report'); |
| | | $message .= sprintf('<p>Report for <strong>%s</strong></p>', $yesterday); |
| | | |
| | | // Summary stats in grid |
| | | $stats = [ |
| | | JVB()->email()->stat($total_ops, 'Operations'), |
| | | JVB()->email()->stat($total_success, 'Successful', '✓'), |
| | | JVB()->email()->stat($total_failed, 'Failed', $total_failed > 0 ? '⚠' : ''), |
| | | JVB()->email()->stat($success_rate . '%', 'Success Rate') |
| | | Emails::getInstance()->stat($total_ops, 'Operations'), |
| | | Emails::getInstance()->stat($total_success, 'Successful', '✓'), |
| | | Emails::getInstance()->stat($total_failed, 'Failed', $total_failed > 0 ? '⚠' : ''), |
| | | Emails::getInstance()->stat($success_rate . '%', 'Success Rate') |
| | | ]; |
| | | $message .= JVB()->email()->grid($stats, 4); |
| | | $message .= Emails::getInstance()->grid($stats, 4); |
| | | |
| | | $message .= JVB()->email()->spacer(20); |
| | | $message .= Emails::getInstance()->spacer(20); |
| | | |
| | | // Alert if success rate is low |
| | | if ($success_rate < 90) { |
| | | $message .= JVB()->email()->alert( |
| | | $message .= Emails::getInstance()->alert( |
| | | sprintf('Success rate of %s%% is below the 90%% threshold', $success_rate), |
| | | 'warning' |
| | | ); |
| | | } |
| | | |
| | | $message .= JVB()->email()->h2('Details by Operation Type'); |
| | | $message .= Emails::getInstance()->h2('Details by Operation Type'); |
| | | |
| | | // Details for each operation type |
| | | foreach ($metrics as $metric) { |
| | | $details = []; |
| | | $details[] = ['label' => 'Total', 'value' => $metric->total_operations]; |
| | | $details[] = ['label' => 'Success', 'value' => JVB()->email()->badge($metric->successful_operations, 'success')]; |
| | | $details[] = ['label' => 'Failed', 'value' => $metric->failed_operations > 0 ? JVB()->email()->badge($metric->failed_operations, 'error') : '0']; |
| | | $details[] = ['label' => 'Success', 'value' => Emails::getInstance()->badge($metric->successful_operations, 'success')]; |
| | | $details[] = ['label' => 'Failed', 'value' => $metric->failed_operations > 0 ? Emails::getInstance()->badge($metric->failed_operations, 'error') : '0']; |
| | | |
| | | if ($metric->average_duration) { |
| | | $details[] = ['label' => 'Avg Duration', 'value' => round($metric->average_duration, 2) . 's']; |
| | |
| | | $details[] = ['label' => 'Peak Memory', 'value' => $memory_mb . ' MB']; |
| | | } |
| | | |
| | | $message .= JVB()->email()->card( |
| | | JVB()->email()->table($details), |
| | | $message .= Emails::getInstance()->card( |
| | | Emails::getInstance()->table($details), |
| | | esc_html($metric->type) |
| | | ); |
| | | } |
| | |
| | | // Current queue status |
| | | $pending_count = $this->getCurrentQueueSize(); |
| | | if ($pending_count > 0) { |
| | | $message .= JVB()->email()->spacer(20); |
| | | $message .= JVB()->email()->notice( |
| | | $message .= Emails::getInstance()->spacer(20); |
| | | $message .= Emails::getInstance()->notice( |
| | | sprintf('<strong>Current Queue:</strong> %d operations pending', $pending_count) |
| | | ); |
| | | } |
| | | |
| | | $message .= JVB()->email()->spacer(20); |
| | | $message .= JVB()->email()->button(admin_url('admin.php?page=jvb-queue'), 'View Queue Dashboard'); |
| | | $message .= Emails::getInstance()->spacer(20); |
| | | $message .= Emails::getInstance()->button(admin_url('admin.php?page=jvb-queue'), 'View Queue Dashboard'); |
| | | |
| | | JVB()->email()->sendEmail($admin_email, $subject, $message, 'QUEUE REPORT'); |
| | | Emails::getInstance()->sendEmail($admin_email, $subject, $message, 'QUEUE REPORT'); |
| | | } |
| | | |
| | | /** |
| | |
| | | SELECT |
| | | status, |
| | | COUNT(*) as count, |
| | | SUM(CASE WHEN status = 'scheduled' AND scheduled_at <= '$current_time' THEN 1 ELSE 0 END) as scheduled_ready |
| | | SUM(IF(status = 'scheduled' AND scheduled_at <= '$current_time', 1, 0)) as scheduled_ready |
| | | FROM $table |
| | | GROUP BY status |
| | | ", OBJECT_K); |
| | |
| | | |
| | | return $value; |
| | | } catch (Exception $e) { |
| | | JVB()->error()->log( |
| | | Errors::getInstance()->log( |
| | | '[OperationQueue]:getOperationValue', |
| | | 'Error retrieving operation value: ' . $e->getMessage(), |
| | | [ |
| | |
| | | DATE(completed_at) as date, |
| | | type, |
| | | COUNT(*) as total_operations, |
| | | SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END) as successful, |
| | | SUM(CASE WHEN status LIKE 'failed%' THEN 1 ELSE 0 END) as failed, |
| | | SUM(IF(status = 'completed', 1, 0)) as successful, |
| | | SUM(IF(status LIKE 'failed%', 1, 0)) as failed, |
| | | AVG(TIMESTAMPDIFF(SECOND, created_at, completed_at)) as avg_duration, |
| | | SUM(count) as items_processed |
| | | FROM $table |