* @copyright Copyright (C) 2026 Moko Consulting. All rights reserved. * @license GNU General Public License version 3 or later; see LICENSE * SPDX-License-Identifier: GPL-3.0-or-later */ namespace Joomla\Component\MokoSuiteCross\Administrator\Helper; defined('_JEXEC') or die; use Joomla\CMS\Component\ComponentHelper; use Joomla\CMS\Factory; use Joomla\CMS\Plugin\PluginHelper; use Joomla\CMS\Uri\Uri; use Joomla\Component\MokoSuiteCross\Administrator\Helper\CredentialHelper; use Joomla\Component\MokoSuiteCross\Administrator\Service\MokoSuiteCrossServiceInterface; /** * Shared queue processor used by: * - System plugin onAfterRender (page-load processing) * - Task scheduler plugin (Joomla scheduled task) * * Handles: queued posts, failed retries, scheduled posts, and log cleanup. * Uses a simple DB-based lock to prevent concurrent execution. */ class QueueProcessor { /** * Process the post queue: dispatch queued posts, retry failed, fire scheduled. * * @param int $batchSize Max posts to process per run * * @return array ['processed' => int, 'succeeded' => int, 'failed' => int, 'skipped' => int] */ public static function processQueue(int $batchSize = 10): array { $result = ['processed' => 0, 'succeeded' => 0, 'failed' => 0, 'skipped' => 0]; if (!self::acquireLock()) { $result['skipped'] = -1; return $result; } try { $db = Factory::getDbo(); $componentParams = ComponentHelper::getParams('com_mokosuitecross'); $maxRetry = (int) $componentParams->get('retry_max', 3); $retryDelay = (int) $componentParams->get('retry_delay', 300); $now = Factory::getDate()->toSql(); // Build service plugin map $pluginMap = self::getServicePluginMap(); // 1. Process queued posts $query = $db->getQuery(true) ->select('p.*, s.service_type, s.credentials, s.params AS service_params') ->from($db->quoteName('#__mokosuitecross_posts', 'p')) ->join('INNER', $db->quoteName('#__mokosuitecross_services', 's') . ' ON ' . $db->quoteName('s.id') . ' = ' . $db->quoteName('p.service_id')) ->where($db->quoteName('p.status') . ' = ' . $db->quote('queued')) ->where('(' . $db->quoteName('p.scheduled_at') . ' IS NULL OR ' . $db->quoteName('p.scheduled_at') . ' <= ' . $db->quote($now) . ')') ->where($db->quoteName('s.published') . ' = 1') ->order($db->quoteName('p.created') . ' ASC') ->setLimit($batchSize); $db->setQuery($query); $queuedPosts = $db->loadObjectList() ?: []; // 2. Process failed posts eligible for retry (exponential backoff) // Retry 1 waits retryDelay, retry 2 waits retryDelay*2, retry 3 waits retryDelay*4, etc. $query = $db->getQuery(true) ->select('p.*, s.service_type, s.credentials, s.params AS service_params') ->from($db->quoteName('#__mokosuitecross_posts', 'p')) ->join('INNER', $db->quoteName('#__mokosuitecross_services', 's') . ' ON ' . $db->quoteName('s.id') . ' = ' . $db->quoteName('p.service_id')) ->where($db->quoteName('p.status') . ' = ' . $db->quote('failed')) ->where($db->quoteName('p.retry_count') . ' < ' . $maxRetry) ->where($db->quoteName('p.modified') . ' <= DATE_SUB(NOW(), INTERVAL (' . (int) $retryDelay . ' * POW(2, ' . $db->quoteName('p.retry_count') . ')) SECOND)') ->where($db->quoteName('s.published') . ' = 1') ->order($db->quoteName('p.modified') . ' ASC') ->setLimit($batchSize); $db->setQuery($query); $retryPosts = $db->loadObjectList() ?: []; // 3. Recover stale "posting" entries (stuck > 10 minutes) $staleQuery = $db->getQuery(true) ->update($db->quoteName('#__mokosuitecross_posts')) ->set($db->quoteName('status') . ' = ' . $db->quote('queued')) ->set($db->quoteName('modified') . ' = ' . $db->quote($now)) ->where($db->quoteName('status') . ' = ' . $db->quote('posting')) ->where($db->quoteName('modified') . ' < DATE_SUB(NOW(), INTERVAL 600 SECOND)'); $db->setQuery($staleQuery); $db->execute(); $allPosts = array_merge($queuedPosts, $retryPosts); foreach ($allPosts as $post) { $result['processed']++; $plugin = $pluginMap[$post->service_type] ?? null; if (!$plugin) { $result['skipped']++; continue; } $isRetry = ($post->status === 'failed'); if ($isRetry) { $newRetryCount = (int) $post->retry_count + 1; // If this is the last retry attempt, mark permanently failed on failure if ($newRetryCount >= $maxRetry) { $db->setQuery( $db->getQuery(true) ->update($db->quoteName('#__mokosuitecross_posts')) ->set($db->quoteName('status') . ' = ' . $db->quote('permanently_failed')) ->set($db->quoteName('retry_count') . ' = ' . $newRetryCount) ->set($db->quoteName('error_message') . ' = CONCAT(' . $db->quoteName('error_message') . ', ' . $db->quote(' [max retries exceeded]') . ')') ->set($db->quoteName('modified') . ' = ' . $db->quote(Factory::getDate()->toSql())) ->where($db->quoteName('id') . ' = ' . (int) $post->id) ); $db->execute(); self::log($db, (int) $post->id, (int) $post->service_id, 'error', sprintf('Permanently failed %s: max retries (%d) exceeded', $post->service_type, $maxRetry)); $result['failed']++; continue; } $db->setQuery( $db->getQuery(true) ->update($db->quoteName('#__mokosuitecross_posts')) ->set($db->quoteName('retry_count') . ' = ' . $newRetryCount) ->where($db->quoteName('id') . ' = ' . (int) $post->id) ); $db->execute(); } // Mark as posting $db->setQuery( $db->getQuery(true) ->update($db->quoteName('#__mokosuitecross_posts')) ->set($db->quoteName('status') . ' = ' . $db->quote('posting')) ->set($db->quoteName('modified') . ' = ' . $db->quote(Factory::getDate()->toSql())) ->where($db->quoteName('id') . ' = ' . (int) $post->id) ); $db->execute(); $credentials = CredentialHelper::decrypt($post->credentials ?: ''); $params = json_decode($post->service_params ?: '{}', true) ?: []; // Token auto-refresh before posting OAuthHelper::refreshTokenIfNeeded((int) $post->service_id, $credentials); // Extract intro image for media attachment $media = []; if (!empty($post->article_id)) { $imgQuery = $db->getQuery(true) ->select($db->quoteName('images')) ->from($db->quoteName('#__content')) ->where($db->quoteName('id') . ' = ' . (int) $post->article_id); $db->setQuery($imgQuery); $imgJson = $db->loadResult(); if ($imgJson) { $imgData = json_decode($imgJson); if (!empty($imgData->image_intro)) { $media[] = Uri::root() . ltrim($imgData->image_intro, '/'); } } } // Lifecycle event: before post $cancel = false; $message = $post->message; try { $dispatcher = Factory::getApplication()->getDispatcher(); $beforeEvent = new \Joomla\Event\Event('onMokoSuiteCrossBeforePost', [(int) $post->id, &$message, $post->service_type, &$cancel]); $dispatcher->dispatch('onMokoSuiteCrossBeforePost', $beforeEvent); } catch (\Throwable $e) { // Dispatcher may not be available } if ($cancel) { $db->setQuery( $db->getQuery(true) ->update($db->quoteName('#__mokosuitecross_posts')) ->set($db->quoteName('status') . ' = ' . $db->quote('cancelled')) ->set($db->quoteName('modified') . ' = ' . $db->quote(Factory::getDate()->toSql())) ->where($db->quoteName('id') . ' = ' . (int) $post->id) ); $db->execute(); self::log($db, (int) $post->id, (int) $post->service_id, 'info', sprintf('Post to %s cancelled by onMokoSuiteCrossBeforePost event', $post->service_type)); $result['skipped']++; continue; } try { $apiResult = $plugin->publish($message, $media, $credentials, $params); if (!empty($apiResult['success'])) { $db->setQuery( $db->getQuery(true) ->update($db->quoteName('#__mokosuitecross_posts')) ->set($db->quoteName('status') . ' = ' . $db->quote('posted')) ->set($db->quoteName('platform_post_id') . ' = ' . $db->quote($apiResult['platform_post_id'] ?? '')) ->set($db->quoteName('platform_response') . ' = ' . $db->quote(json_encode($apiResult['response'] ?? []))) ->set($db->quoteName('posted_at') . ' = ' . $db->quote(Factory::getDate()->toSql())) ->set($db->quoteName('modified') . ' = ' . $db->quote(Factory::getDate()->toSql())) ->where($db->quoteName('id') . ' = ' . (int) $post->id) ); $db->execute(); self::log($db, (int) $post->id, (int) $post->service_id, 'info', sprintf('%s to %s (ID: %s)', $isRetry ? 'Retry succeeded' : 'Posted', $post->service_type, $apiResult['platform_post_id'] ?? 'n/a')); // Lifecycle event: after successful post try { $afterEvent = new \Joomla\Event\Event('onMokoSuiteCrossAfterPost', [(int) $post->id, $post->service_type, $apiResult]); $dispatcher->dispatch('onMokoSuiteCrossAfterPost', $afterEvent); } catch (\Throwable $e) { // Non-critical } $result['succeeded']++; } else { $errorMsg = $apiResult['response']['error'] ?? json_encode($apiResult['response'] ?? []); $db->setQuery( $db->getQuery(true) ->update($db->quoteName('#__mokosuitecross_posts')) ->set($db->quoteName('status') . ' = ' . $db->quote('failed')) ->set($db->quoteName('error_message') . ' = ' . $db->quote(mb_substr($errorMsg, 0, 1000))) ->set($db->quoteName('platform_response') . ' = ' . $db->quote(json_encode($apiResult['response'] ?? []))) ->set($db->quoteName('modified') . ' = ' . $db->quote(Factory::getDate()->toSql())) ->where($db->quoteName('id') . ' = ' . (int) $post->id) ); $db->execute(); self::log($db, (int) $post->id, (int) $post->service_id, 'error', sprintf('Failed %s: %s', $post->service_type, mb_substr($errorMsg, 0, 500))); // Lifecycle event: post failed try { $failedEvent = new \Joomla\Event\Event('onMokoSuiteCrossPostFailed', [(int) $post->id, $post->service_type, $errorMsg]); $dispatcher->dispatch('onMokoSuiteCrossPostFailed', $failedEvent); } catch (\Throwable $e) { // Non-critical } $result['failed']++; } } catch (\Throwable $e) { $db->setQuery( $db->getQuery(true) ->update($db->quoteName('#__mokosuitecross_posts')) ->set($db->quoteName('status') . ' = ' . $db->quote('failed')) ->set($db->quoteName('error_message') . ' = ' . $db->quote(mb_substr($e->getMessage(), 0, 1000))) ->set($db->quoteName('modified') . ' = ' . $db->quote(Factory::getDate()->toSql())) ->where($db->quoteName('id') . ' = ' . (int) $post->id) ); $db->execute(); self::log($db, (int) $post->id, (int) $post->service_id, 'error', sprintf('Exception %s: %s', $post->service_type, mb_substr($e->getMessage(), 0, 500))); // Lifecycle event: post failed (exception) try { $failedEvent = new \Joomla\Event\Event('onMokoSuiteCrossPostFailed', [(int) $post->id, $post->service_type, $e->getMessage()]); $dispatcher->dispatch('onMokoSuiteCrossPostFailed', $failedEvent); } catch (\Throwable $ex) { // Non-critical } $result['failed']++; } } // 3. Clean up old logs self::cleanupLogs($db, $componentParams); } finally { self::releaseLock(); } return $result; } /** * Process evergreen re-shares: find articles marked as evergreen whose last * successful post to each service was longer ago than the configured interval, * and create new queue entries for them. * * @return array ['queued' => int] */ public static function processEvergreen(): array { $result = ['queued' => 0]; $componentParams = ComponentHelper::getParams('com_mokosuitecross'); if (!$componentParams->get('evergreen_enabled', 1)) { return $result; } $defaultInterval = (int) $componentParams->get('evergreen_default_interval', 30); $maxPerRun = (int) $componentParams->get('evergreen_max_per_run', 3); $db = Factory::getDbo(); $now = Factory::getDate()->toSql(); // Find published articles with evergreen=1 in attribs $query = $db->getQuery(true) ->select('c.id, c.attribs') ->from($db->quoteName('#__content', 'c')) ->where($db->quoteName('c.state') . ' = 1') ->where('JSON_EXTRACT(' . $db->quoteName('c.attribs') . ', ' . $db->quote('$.mokosuitecross_evergreen') . ') = ' . $db->quote('1')); $db->setQuery($query); $articles = $db->loadObjectList() ?: []; if (empty($articles)) { return $result; } // Load all published services $query = $db->getQuery(true) ->select('id, service_type') ->from($db->quoteName('#__mokosuitecross_services')) ->where($db->quoteName('published') . ' = 1'); $db->setQuery($query); $services = $db->loadObjectList() ?: []; if (empty($services)) { return $result; } // Import service plugins (not used for direct dispatch here, but ensures // they are loaded in case any lifecycle events depend on them) PluginHelper::importPlugin('mokosuitecross'); // Batch pre-load: latest posted_at per article+service (eliminates N*M queries) $articleIds = implode(',', array_map(function ($a) { return (int) $a->id; }, $articles)); $serviceIds = implode(',', array_map(function ($s) { return (int) $s->id; }, $services)); $query = $db->getQuery(true) ->select(['article_id', 'service_id', 'MAX(' . $db->quoteName('posted_at') . ') AS last_posted']) ->from($db->quoteName('#__mokosuitecross_posts')) ->where($db->quoteName('article_id') . ' IN (' . $articleIds . ')') ->where($db->quoteName('service_id') . ' IN (' . $serviceIds . ')') ->where($db->quoteName('status') . ' = ' . $db->quote('posted')) ->group(['article_id', 'service_id']); $db->setQuery($query); $lastPostedRows = $db->loadObjectList() ?: []; $lastPostedMap = []; foreach ($lastPostedRows as $row) { $lastPostedMap[$row->article_id . ':' . $row->service_id] = $row->last_posted; } // Batch pre-load: existing queued/posting entries $query = $db->getQuery(true) ->select(['article_id', 'service_id']) ->from($db->quoteName('#__mokosuitecross_posts')) ->where($db->quoteName('article_id') . ' IN (' . $articleIds . ')') ->where($db->quoteName('service_id') . ' IN (' . $serviceIds . ')') ->where($db->quoteName('status') . ' IN (' . $db->quote('queued') . ',' . $db->quote('posting') . ')'); $db->setQuery($query); $pendingRows = $db->loadObjectList() ?: []; $pendingSet = []; foreach ($pendingRows as $row) { $pendingSet[$row->article_id . ':' . $row->service_id] = true; } foreach ($articles as $article) { if ($result['queued'] >= $maxPerRun) { break; } $attribs = json_decode($article->attribs ?? '{}', true) ?: []; $interval = (int) ($attribs['mokosuitecross_evergreen_interval'] ?? $defaultInterval); if ($interval < 1) { $interval = $defaultInterval; } // Per-article service filter $selectedServiceIds = $attribs['mokosuitecross_services'] ?? null; if (is_array($selectedServiceIds) && !empty($selectedServiceIds)) { $selectedServiceIds = array_map('intval', $selectedServiceIds); } else { $selectedServiceIds = null; } // Load the full article for template rendering $fullArticle = null; foreach ($services as $service) { if ($result['queued'] >= $maxPerRun) { break; } // Per-article service filter if ($selectedServiceIds !== null && !in_array((int) $service->id, $selectedServiceIds, true)) { continue; } $key = $article->id . ':' . $service->id; // Check last successful post from batch-loaded map $lastPosted = $lastPostedMap[$key] ?? null; if (empty($lastPosted)) { // Never posted — skip, the initial cross-post will handle it continue; } // Check if interval has elapsed $dueDate = Factory::getDate($lastPosted . ' + ' . $interval . ' days'); if ($dueDate->toUnix() > Factory::getDate()->toUnix()) { continue; } // Skip if there's already a queued/posting entry if (isset($pendingSet[$key])) { continue; } // Load full article if not already loaded if ($fullArticle === null) { $query = $db->getQuery(true) ->select('*') ->from($db->quoteName('#__content')) ->where($db->quoteName('id') . ' = ' . (int) $article->id); $db->setQuery($query); $fullArticle = $db->loadObject(); if (!$fullArticle) { break; } } // Render message using default template $template = $componentParams->get('default_template', "{title}\n\n{url}"); $message = self::renderEvergreenMessage($db, $fullArticle, $template); // Create queue entry $post = (object) [ 'article_id' => (int) $article->id, 'service_id' => (int) $service->id, 'status' => 'queued', 'message' => $message, 'platform_post_id' => '', 'platform_response' => '', 'error_message' => '', 'retry_count' => 0, 'created' => $now, 'modified' => $now, ]; $db->insertObject('#__mokosuitecross_posts', $post); self::log($db, $db->insertid(), (int) $service->id, 'info', sprintf('Evergreen re-share queued for article %d to %s (interval: %d days)', $article->id, $service->service_type, $interval)); $result['queued']++; } } return $result; } /** * Render a message for an evergreen re-share using the default template. */ private static function renderEvergreenMessage($db, object $article, string $template): string { $url = \Joomla\CMS\Uri\Uri::root() . 'index.php?option=com_content&view=article&id=' . $article->id; if (!empty($article->catid)) { $url .= '&catid=' . $article->catid; } $categoryName = ''; if (!empty($article->catid)) { $query = $db->getQuery(true) ->select($db->quoteName('title')) ->from($db->quoteName('#__categories')) ->where($db->quoteName('id') . ' = ' . (int) $article->catid); $db->setQuery($query); $categoryName = $db->loadResult() ?: ''; } $authorName = ''; if (!empty($article->created_by)) { $query = $db->getQuery(true) ->select($db->quoteName('name')) ->from($db->quoteName('#__users')) ->where($db->quoteName('id') . ' = ' . (int) $article->created_by); $db->setQuery($query); $authorName = $db->loadResult() ?: ''; } $introImage = ''; $images = json_decode($article->images ?? '{}'); if (!empty($images->image_intro)) { $introImage = \Joomla\CMS\Uri\Uri::root() . ltrim($images->image_intro, '/'); } // Resolve article tags $tagNames = []; if (!empty($article->id)) { $query = $db->getQuery(true) ->select($db->quoteName('t.title')) ->from($db->quoteName('#__tags', 't')) ->join('INNER', $db->quoteName('#__contentitem_tag_map', 'm') . ' ON ' . $db->quoteName('m.tag_id') . ' = ' . $db->quoteName('t.id')) ->where($db->quoteName('m.type_alias') . ' = ' . $db->quote('com_content.article')) ->where($db->quoteName('m.content_item_id') . ' = ' . (int) $article->id) ->where($db->quoteName('t.published') . ' = 1'); $db->setQuery($query); $tagNames = $db->loadColumn() ?: []; } $tagsComma = implode(', ', $tagNames); $hashtags = implode(' ', array_map(function ($tag) { return '#' . preg_replace('/\s+/', '', $tag); }, $tagNames)); $replacements = [ '{title}' => $article->title ?? '', '{introtext}' => strip_tags(mb_substr($article->introtext ?? '', 0, 280)), '{fulltext}' => strip_tags(mb_substr($article->fulltext ?? '', 0, 500)), '{url}' => $url, '{image}' => $introImage, '{category}' => $categoryName, '{author}' => $authorName, '{date}' => Factory::getDate($article->publish_up ?? 'now')->format('Y-m-d'), '{tags}' => $tagsComma, '{hashtags}' => $hashtags, ]; $message = str_replace(array_keys($replacements), array_values($replacements), $template); // Resolve custom field placeholders: {field:field_name} $message = preg_replace_callback('/\{field:([a-zA-Z0-9_-]+)\}/', function ($matches) use ($db, $article) { $fieldName = $matches[1]; $query = $db->getQuery(true) ->select('fv.value') ->from($db->quoteName('#__fields_values', 'fv')) ->join('INNER', $db->quoteName('#__fields', 'f') . ' ON f.id = fv.field_id') ->where('f.name = ' . $db->quote($fieldName)) ->where('fv.item_id = ' . (int) $article->id); $db->setQuery($query); return $db->loadResult() ?: ''; }, $message); return $message; } /** * Manually retry one or more failed/permanently_failed posts. * * Resets status to 'queued' and retry_count to 0 so the queue processor * picks them up on the next run. * * @param array $postIds Post IDs to retry * * @return int Number of posts re-queued */ public static function retryPosts(array $postIds): int { if (empty($postIds)) { return 0; } $db = Factory::getDbo(); $now = Factory::getDate()->toSql(); $ids = implode(',', array_map('intval', $postIds)); $query = $db->getQuery(true) ->update($db->quoteName('#__mokosuitecross_posts')) ->set($db->quoteName('status') . ' = ' . $db->quote('queued')) ->set($db->quoteName('retry_count') . ' = 0') ->set($db->quoteName('error_message') . ' = ' . $db->quote('')) ->set($db->quoteName('modified') . ' = ' . $db->quote($now)) ->where($db->quoteName('id') . ' IN (' . $ids . ')') ->where($db->quoteName('status') . ' IN (' . $db->quote('failed') . ',' . $db->quote('permanently_failed') . ')'); $db->setQuery($query); $db->execute(); $count = $db->getAffectedRows(); if ($count > 0) { self::log($db, null, null, 'info', sprintf('Manual retry: %d post(s) re-queued', $count)); } return $count; } /** * Retry all failed posts for a specific service. * * @param int $serviceId Service ID * * @return int Number of posts re-queued */ public static function retryService(int $serviceId): int { $db = Factory::getDbo(); $now = Factory::getDate()->toSql(); $query = $db->getQuery(true) ->update($db->quoteName('#__mokosuitecross_posts')) ->set($db->quoteName('status') . ' = ' . $db->quote('queued')) ->set($db->quoteName('retry_count') . ' = 0') ->set($db->quoteName('error_message') . ' = ' . $db->quote('')) ->set($db->quoteName('modified') . ' = ' . $db->quote($now)) ->where($db->quoteName('service_id') . ' = ' . $serviceId) ->where($db->quoteName('status') . ' IN (' . $db->quote('failed') . ',' . $db->quote('permanently_failed') . ')'); $db->setQuery($query); $db->execute(); $count = $db->getAffectedRows(); if ($count > 0) { self::log($db, null, $serviceId, 'info', sprintf('Bulk retry: %d post(s) re-queued for service %d', $count, $serviceId)); } return $count; } /** * Check if there are pending items in the queue. * * @return bool */ public static function hasPendingWork(): bool { $db = Factory::getDbo(); $componentParams = ComponentHelper::getParams('com_mokosuitecross'); $maxRetry = (int) $componentParams->get('retry_max', 3); $retryDelay = (int) $componentParams->get('retry_delay', 300); $now = Factory::getDate()->toSql(); // Queued posts ready to go $query = $db->getQuery(true) ->select('COUNT(*)') ->from($db->quoteName('#__mokosuitecross_posts')) ->where($db->quoteName('status') . ' = ' . $db->quote('queued')) ->where('(' . $db->quoteName('scheduled_at') . ' IS NULL OR ' . $db->quoteName('scheduled_at') . ' <= ' . $db->quote($now) . ')'); $db->setQuery($query); $queued = (int) $db->loadResult(); // Failed posts eligible for retry (exponential backoff matching processQueue) $query = $db->getQuery(true) ->select('COUNT(*)') ->from($db->quoteName('#__mokosuitecross_posts')) ->where($db->quoteName('status') . ' = ' . $db->quote('failed')) ->where($db->quoteName('retry_count') . ' < ' . $maxRetry) ->where($db->quoteName('modified') . ' <= DATE_SUB(NOW(), INTERVAL (' . (int) $retryDelay . ' * POW(2, ' . $db->quoteName('retry_count') . ')) SECOND)'); $db->setQuery($query); $retryable = (int) $db->loadResult(); return ($queued + $retryable) > 0; } /** * Import mokosuitecross plugins and build a type → plugin instance map. * * @return array */ private static function getServicePluginMap(): array { PluginHelper::importPlugin('mokosuitecross'); // In Joomla 5+ with SubscriberInterface, plugins receive the Event object // as their first argument. When they do $services[] = $this, they append to // the Event via ArrayAccess at numeric indices starting at 1. $servicePlugins = []; $event = new \Joomla\Event\Event('onMokoSuiteCrossGetServices', [$servicePlugins]); try { Factory::getApplication()->getDispatcher()->dispatch( 'onMokoSuiteCrossGetServices', $event ); } catch (\Throwable $e) { // Dispatcher may not be available in all contexts } // Read plugins back from the Event's ArrayAccess indices $idx = 1; while (isset($event[$idx])) { $servicePlugins[] = $event[$idx]; $idx++; } $map = []; foreach ($servicePlugins as $plugin) { if ($plugin instanceof MokoSuiteCrossServiceInterface) { $map[$plugin->getServiceType()] = $plugin; } } return $map; } /** * Delete logs older than the configured retention period. */ private static function cleanupLogs($db, $componentParams): void { $retentionDays = (int) $componentParams->get('log_retention_days', 90); if ($retentionDays <= 0) { return; } $cutoff = Factory::getDate('now - ' . $retentionDays . ' days')->toSql(); $query = $db->getQuery(true) ->delete($db->quoteName('#__mokosuitecross_logs')) ->where($db->quoteName('created') . ' < ' . $db->quote($cutoff)); $db->setQuery($query); $db->execute(); } /** * Acquire a database lock to prevent concurrent queue processing. * * Uses MySQL GET_LOCK() or PostgreSQL pg_advisory_lock() when available, * falling back to a timestamp-based check for other databases. */ private static function acquireLock(): bool { $db = Factory::getDbo(); try { $serverType = $db->getServerType(); if ($serverType === 'mysql' || $serverType === 'mariadb') { $db->setQuery("SELECT GET_LOCK('mokosuitecross_queue', 0)"); return (int) $db->loadResult() === 1; } if ($serverType === 'postgresql') { $db->setQuery("SELECT pg_try_advisory_lock(hashtext('mokosuitecross_queue'))"); return (bool) $db->loadResult(); } } catch (\Throwable $e) { // Fall through to timestamp-based lock } return self::acquireTimestampLock($db); } /** * Release the database lock. */ private static function releaseLock(): void { $db = Factory::getDbo(); try { $serverType = $db->getServerType(); if ($serverType === 'mysql' || $serverType === 'mariadb') { $db->setQuery("SELECT RELEASE_LOCK('mokosuitecross_queue')"); $db->execute(); return; } if ($serverType === 'postgresql') { $db->setQuery("SELECT pg_advisory_unlock(hashtext('mokosuitecross_queue'))"); $db->execute(); return; } } catch (\Throwable $e) { // Fall through to timestamp-based release } self::releaseTimestampLock($db); } /** * Timestamp-based lock fallback for databases without advisory locks. * * Uses an atomic UPDATE with a WHERE clause to prevent TOCTOU race * conditions. The lock is considered stale after 120 seconds. */ private static function acquireTimestampLock($db): bool { $now = time(); $staleThreshold = $now - 120; // Atomic: only succeeds if lock is absent (0) or stale $params = ComponentHelper::getParams('com_mokosuitecross'); $oldParams = $params->toString(); $params->set('queue_lock_time', $now); $newParams = $params->toString(); $query = $db->getQuery(true) ->update($db->quoteName('#__extensions')) ->set($db->quoteName('params') . ' = ' . $db->quote($newParams)) ->where($db->quoteName('element') . ' = ' . $db->quote('com_mokosuitecross')) ->where($db->quoteName('type') . ' = ' . $db->quote('component')) ->where('(' . $db->quoteName('params') . ' NOT LIKE ' . $db->quote('%"queue_lock_time"%') . ' OR ' . $db->quoteName('params') . ' LIKE ' . $db->quote('%"queue_lock_time":0%') . ' OR ' . $db->quoteName('params') . ' LIKE ' . $db->quote('%"queue_lock_time":"0"%') . ')'); $db->setQuery($query); $db->execute(); if ($db->getAffectedRows() > 0) { return true; } // Check if the existing lock is stale $params = ComponentHelper::getParams('com_mokosuitecross'); $lockTime = (int) $params->get('queue_lock_time', 0); if ($lockTime > 0 && $lockTime <= $staleThreshold) { // Force acquire stale lock $params->set('queue_lock_time', $now); $query = $db->getQuery(true) ->update($db->quoteName('#__extensions')) ->set($db->quoteName('params') . ' = ' . $db->quote($params->toString())) ->where($db->quoteName('element') . ' = ' . $db->quote('com_mokosuitecross')) ->where($db->quoteName('type') . ' = ' . $db->quote('component')); $db->setQuery($query); $db->execute(); return true; } return false; } /** * Release the timestamp-based lock. */ private static function releaseTimestampLock($db): void { $params = ComponentHelper::getParams('com_mokosuitecross'); $params->set('queue_lock_time', 0); $query = $db->getQuery(true) ->update($db->quoteName('#__extensions')) ->set($db->quoteName('params') . ' = ' . $db->quote($params->toString())) ->where($db->quoteName('element') . ' = ' . $db->quote('com_mokosuitecross')) ->where($db->quoteName('type') . ' = ' . $db->quote('component')); $db->setQuery($query); $db->execute(); } /** * Write a log entry. */ private static function log($db, ?int $postId, ?int $serviceId, string $level, string $message): void { $log = (object) [ 'post_id' => $postId, 'service_id' => $serviceId, 'level' => $level, 'message' => mb_substr($message, 0, 2000), 'context' => '{}', 'created' => Factory::getDate()->toSql(), ]; $db->insertObject('#__mokosuitecross_logs', $log); } }