Clusterify.AI

Asynchronous Catalog Pipeline Specification

Magento 2 ChatBot & Assistant RabbitMQ Sync & Indexing Architecture

How the Clusterify.AI Magento 2 extension synchronizes tens of thousands of catalog products, categories, and CMS pages without locking database tables or consuming storefront web workers.

The 4-Stage Asynchronous Pipeline

To guarantee that large imports, flash sales, and mass attribute updates never degrade customer browsing or checkout performance, the extension strictly decouples change detection from external API communication.

STAGE 1: DETECTION

Dual-Layer Change Tracking

Database changelog triggers (Mview) pair with save-commit observers to immediately capture product, category, and CMS updates.

STAGE 2: PRODUCING

100-Item Chunking

Three dedicated indexers (CMS, Categories, Products) chunk unique positive entity IDs in batches of 100, publishing to AMQP with zero external network wait.

STAGE 3: BUFFERING

RabbitMQ Queues

Dedicated AMQP message queues safely buffer synchronization workloads without consuming PHP-FPM web worker threads.

STAGE 4: CONSUMPTION

Smart Consumers

Background consumers load entities, convert HTML to clean Markdown, enforce 300ms throttling, and handle rate limits gracefully.

Catalog Knowledge Vectorization & Embedding PipelineSub-60ms Retrieval

Once ingested by api.clusterify.ai, catalog Markdown payloads undergo automated Catalog Knowledge Vectorization. Products, configurable variant options, and custom AI attributes are partitioned into semantic fragments and transformed into high-dimensional vector embeddings. These embeddings power our sub-60ms vector search engine, ensuring your Magento chatbot generates grounded, accurate product recommendations with near-zero hallucination.

Unified Short-Circuit Guard (PlanService::canSyncEntity)

To prevent redundant processing and completely eliminate unnecessary database queries or external API calls, all synchronization touchpoints (Indexers, Queue Consumers, Observers, and CLI runners) evaluate a single authoritative guard:

PlanService::canSyncEntity(string $entityType, ?int $storeId): bool
Step 1 (Zero Network): Checks master toggle (isEnabled). If disabled, returns false immediately.
Step 2 (Zero Network): Checks URL Knowledge Base sync toggle (isSyncEnabled). If disabled, returns false immediately.
Step 3 (Zero Network): Checks entity toggle (isCmsSyncEnabled, isCategorySyncEnabled, isProductSyncEnabled). If disabled, returns false.
Step 4 (Zero Network): Checks that API keys are configured (getPublicKey, getSecretKey). If missing, returns false.
Step 5 (Plan Gate): Only if all 4 local checks pass does it verify plan eligibility (isUrlKnowledgeAllowed), cached for 10 minutes.

Entity Lifecycle & State Handling Rules

The synchronization engine handles all lifecycle states automatically, guaranteeing that deleted, disabled, or out-of-stock items never consume URL quotas:

Item Added

upsert(url, content, isEnabled=true)
Trigger: New CMS page, category, or product created & enabled
Action: Extracts metadata, converts HTML to Markdown

Item Updated

upsert(url, updatedContent, isEnabled=true)
Trigger: Description, price, stock, or AI knowledge modified
Action: Re-generates Markdown with updated details

Item Disabled

bulkDelete(urls: [url])
Trigger: Status set to Disabled (status = 2 / is_active = 0)
Action: Detects disabled flag; emits delete action

Item Hidden

bulkDelete(urls: [url])
Trigger: Product visibility changed to Not Visible Individually
Action: Detects hidden state; emits delete action

Item Deleted

bulkDelete(urls: [url])
Trigger: Entity completely removed from Magento database
Action: Pre-deletion observer dispatches delete action

Out of Stock

bulkDelete(urls: [url])
Trigger: in_stock_only = 1 enabled and item goes out of stock
Action: Detects out-of-stock state; purges URL from quota

Consumer Execution Workflows

Pending RabbitMQ tasks can be processed across four execution modes to fit any infrastructure:

1. Automated Magento Cron

Runs clusterify_chatbot_sync_process_queue automatically every minute via standard cron.

2. Dedicated On-Demand CLI

bin/magento clusterify:chatbot:sync:consume --limit=50 immediately drains tasks and exits cleanly.

3. Admin Dashboard Button

Click ⚡ Process Pending Tasks Now on the Admin Status Dashboard for an instant AJAX drain.

4. Persistent Daemon Workers

Run continuous consumers via bin/magento queue:consumers:start managed by Supervisor or Systemd.

Catalog Sync FAQ

Frequently Asked Questions: RabbitMQ Catalog Sync

Technical answers on indexers, rate-limit throttling, lifecycle actions, and queue workers.

Why does the extension use RabbitMQ instead of direct HTTP calls during product saves?
To protect customer checkout and admin saving performance. Direct HTTP requests during entity saves tie up PHP-FPM worker threads and can cause web timeouts. By publishing lightweight messages into RabbitMQ in batches of 100, save operations complete in milliseconds while background consumers handle API communication asynchronously.
How do the 3 dedicated indexers chunk catalog entity updates?
The indexers (clusterify_chatbot_cms, clusterify_chatbot_category, and clusterify_chatbot_product) deduplicate positive entity IDs from Mview changelog tables and publish them to RabbitMQ in manageable batches of 100 messages.
How does the built-in 300ms throttling protect our store against API rate limits?
The abstract consumer enforces a strict 300ms sleep delay between consecutive dispatches to api.clusterify.ai. Combined with network round-trip latency (~200–300ms), this guarantees total throughput remains safely below the 120 req/min API quota, preventing rate limit spikes during bulk catalog indexing.
What happens if an HTTP 429 Rate Limit Exceeded is returned by the API?
The consumer catches RateLimitExceededException from the PHP SDK, dynamically parses $e->getRetryAfter(), logs the event, pauses execution, and retries safely without losing queue state or dropping messages.
How does the 5-step unified short-circuit guard save database and server resources?
Before executing any collection query, all indexers and queue consumers call PlanService::canSyncEntity(). It evaluates 4 local checks (master toggle, sync toggle, entity toggle, API keys) before verifying plan eligibility. If disabled or on the Starter plan, it exits with zero database queries and zero network overhead.
How does the dedicated AI ChatBot Knowledge attribute get synchronized?
The synchronization engine extracts each product's title, SKU, pricing, availability, and configurable options, pairing them with the dedicated 20,000-character AI ChatBot Knowledge attribute. This feeds tailored sales arguments, sizing nuances, and cross-sell pairings directly into the AI context without altering public storefront HTML descriptions. If custom knowledge is not populated for an item, standard storefront descriptions are automatically used. Review our Knowledge Architecture guide.
How are canonical URLs generated for products with category prefixes in their storefront paths?
The product data provider forces _ignore_category => true on the URL model. This mirrors Magento's native canonical tag generation in the HTML head, stripping all category folder prefixes and ensuring each product has a single authoritative knowledge base URL.
How can store operations monitor RabbitMQ queue backlogs in real time?
In the Magento Admin Panel under CHATBOT > Dashboard & Status, a live table displays real-time pending message counts across CMS, Category, and Product queues. An on-demand ⚡ Process Pending Tasks Now button allows instant draining without terminal access. Read our Admin Operations Guide.
How can DevOps engineers trigger or drain consumers from the terminal or CI/CD?
Use bin/magento clusterify:chatbot:sync:run to queue entities, and bin/magento clusterify:chatbot:sync:consume --limit=50 to immediately drain tasks and exit cleanly. Consult our CLI Reference.
Does the synchronization engine support Adobe Commerce Content Staging campaigns?
Yes. Entity providers load items through official Magento repository interfaces plugged by Magento_CatalogStaging and Magento_CmsStaging. Active staged versions are resolved via MetadataPool. Learn more in our Cloud Compatibility Guide.
← Back to Magento 2 OverviewRequest Architecture Review