Queues & Background Processing¶

Most Prism features are initiated by an HTTP request but completed asynchronously. This section documents the queueing model that does the "completed asynchronously" half. If you have not yet read the Request Lifecycle, start there — it frames the sync/async boundary this section drills into.

Validation legend¶

Symbol	Meaning
✅	Verified against source in this review
📄	Documented in the architecture map (`map-04-crons-queues`)
💻	Behaviour confirmed by reading the implementation file
♻️	Legacy / superseded but still running
⚰️	Dead / commented-out code
⚠️	Tech-debt or operational risk — see Technical Debt

1. The core idea: queues are collections, crons are the broker¶

Prism has no external message broker in the monolith. There is no RabbitMQ, no SQS, no Redis Streams, and no Kafka inside uengage-crm. ⚠️ (Kafka exists only in the separate prism-services estate; the monolith documented here does not use it.)

Instead, the pattern is uniform:

A producer (an HTTP controller, another cron, or a helper in commonFunctions/) inserts a row/document into a "queue" — which is just an ordinary MongoDB collection or MySQL table — with a status field marking it not yet processed.
A consumer is an in-process node-cron job (node-cron ^3.0.0) that wakes on a fixed cadence, polls the queue for unprocessed rows, drains a batch, does the work, and flips the status. 💻

flowchart LR
    P["Producer<br/>(controller / helper / cron)"] -->|insert status=pending| Q[("Queue collection/table<br/>Mongo or MySQL")]
    C["node-cron consumer<br/>(polls on a schedule)"] -->|SELECT/find pending| Q
    C -->|do work| EXT["SMS / WhatsApp / Firebase /<br/>Wallet / Milestone"]
    C -->|UPDATE status=done/failed| Q

Every cron require()s itself at startup from index.js and self-registers via cron.schedule(...). There is one process; the crons share it. Crons use the MySQL jobsPool (not the web pool) so background work does not starve web requests — see Caching / Pooling.

Why this matters operationally: when "nothing happened," the answer is almost always a stuck row in a queue collection or a stuck cron. See §6.

2. Generic queue state machine¶

All Prism queues implement the same lifecycle, but ⚠️ the field names and values are inconsistent across queues (see the table in §3). The abstract machine:

stateDiagram-v2
    [*] --> queued: producer inserts (pending / status=0 / processed=false)
    queued --> in_progress: cron picks batch (marks picked where supported)
    in_progress --> done: success (done / processed=true / status=3 sent)
    in_progress --> retry: transient failure (retry_count++, set next_retry_at)
    retry --> in_progress: next tick, once next_retry_at <= now (backoff)
    retry --> failed: retry_count >= MAX_RETRIES
    in_progress --> failed: permanent/validation error
    done --> [*]
    failed --> [*]

Key mechanics that recur:

retry_count + next_retry_at — the richest implementation is crm_points_allocation_queue (MySQL): on a transient miss (e.g. the order document has not yet landed in Mongo) it does retry_count = retry_count + 1, next_retry_at = DATE_ADD(NOW(), INTERVAL 15 MINUTE), and only marks failed once retry_count >= 5. ✅💻
Batch draining — consumers pull a bounded batch per tick (LIMIT 5000 for points, batches of 8000 for SMS) rather than the whole backlog. ✅📄
No global Dead Letter Queue. ⚠️ A failed row simply sits in the table; there is no DLQ and no automatic replay. Recovery is manual or via cleanup crons.
Cron overlap lock — before draining, most crons check a status/lock document (cron_statuses, sms_cron_status, whatsapp_cron_status, push_cron_status) for a prior run still marked status: false and skip if found. ✅💻

3. The real queues¶

Queue	Store	Status field & values	Cron (consumer)	Cadence	Batch / retries
`crm_queue` (orders)	MongoDB	`status` (`null`/pending → `1` processed); `source` 1=Swiggy 2=Zomato 3=POS 4=Dashboard 5=Manual 6=External webhook	ingestion path + `deleteCrmQueue.js` cleanup	frequent poll	—
`crm_points_allocation_queue`	MySQL	`status` `'pending'` \| `'done'` \| `'failed'`; `processed` 0/1	`crmPointsAllocationQueueCron.js`	`/2 * * *` (2 min)	`LIMIT 5000`; 5 retries, 15-min backoff via `next_retry_at`
`loyalty_points_allocate` ♻️	MongoDB	`status` 0=pending 1=done -1=error	`pointAllocation.js`	`/1 * * *` (1 min)	legacy; superseded by the MySQL queue
`whatsapp_queue_YYYYMMDD` (marketing)	MongoDB	`process_status` 0=pending 2=picked 3=sent/pending-webhook 4=skipped; `type` absent	`sendWhatsappMessageCron.js`	~1–3 s poll	groups per send
`whatsapp_queue_YYYYMMDD` (transactional)	MongoDB	same; `type` present	`sendWhatsappTransactionalCron.js`	~1 s poll ⚠️	order-related msgs
`sms_queue_YYYYMMDD`	MongoDB	`processed` bool (false→true) + `execution_time <= now`	`smsQueueCron.js`	`/3 * * *` (3 min)	batches of 8000
`push_noti_queue_YYYYMMDD`	MongoDB	`processed` bool; grouped by `(businessId, os, title, message, campaign_id, fcm_token)`	`pushQueueCron.js`	`/5 * * *` (5 min)	grouped Firebase sends
`notiJunk`	MongoDB	processed/junk markers	`processNotiQueue` + delete crons	periodic	notification overflow / dead records

Date-partitioned collections ⚠️ — SMS, Push and WhatsApp queues are suffixed with YYYYMMDD (producers usually write tomorrow's collection). This spreads load but means every query must know the correct date, and stale collections need a cleanup strategy. See Technical Debt.

3.1 `crm_points_allocation_queue` in detail ✅💻¶

The most important — and most complex — queue. Producer is order ingestion (processCommonIngestion.js); consumer is crmPointsAllocationQueueCron.js (every 2 min, LIMIT 5000). It multiplexes on a type column:

`type`	Handler	Effect
`uen_order`	fraud check → order lookup → `handleWalletPointsFlow()`	credits earned points, sets `addo_orders.cashbackReceived=1`
`order_reward`	`handleOrderRewards()` (from `helpers.js`)	order-driven reward grant
`welcome_point`	wallet-rules lookup → wallet insert/update	one-time welcome bonus (guarded by `customers.welcome_points=1`)
`refer_earn` / `refer*`	wallet insert/update	referral points

Retry logic (verified in source): for uen_order/order_reward, if the matching order document is not yet in Mongo, the row is not failed — it schedules next_retry_at = NOW() + 15 min and increments retry_count, giving up (mark failed) only after 5 attempts (~ up to a couple of hours). Validation errors (bad mobile, blocked customer, missing wallet rules, missing contact mapping) fail immediately with an error_message.

⚠️ Non-atomic across stores. A single row touches MySQL (wallet, wallet_history, wallet_rules) and MongoDB (customers, orders) with no distributed transaction. A crash mid-flow can leave wallet updated but the queue row not marked, or vice versa. Idempotency is only partially enforced (e.g. the welcome_points flag). See Technical Debt.

4. Producers & helpers (`commonFunctions/`, `processLibrary/`)¶

Queues do not fill themselves. The main producers:

Producer	Writes to	Notes
`processCommonIngestion.js`	`orders`, `crm_queue`, `crm_points_allocation_queue`	order ingestion from Swiggy/Zomato/POS/dashboard
`automatedCampaignQueueNew.js` ✅	`sms_queue_`, `whatsapp_queue_`, `push_noti_queue_*`	main automated-campaign builder; validates customer/wallet/opt-in, substitutes template variables, deducts wallet balance
`automatedCampaignQueue.js` ♻️⚠️	same	legacy duplicate of the above — see Cron Catalog tech-debt
`smsCampaignQueue.js` / `pushCampaignQueue.js`	`sms_queue_` / `push_noti_queue_`	thin insert wrappers (write tomorrow's collection)
`uengageProcess.js` / `processCampaign.js`	all three channel queues + `auto_campaign_roi`	large journey/campaign orchestrators
`abandonCartCron.js`	`sms_queue_`, `whatsapp_queue_`	abandoned-cart re-engagement

Consumers of note: whatsappMsgProcess.js (WhatsApp send via Gupshup/Facebook), processNotiQueue.js (Firebase Admin push), sendAutoCampaignSms.js (SMS send). Full inventory in the Cron Catalog.

5. The four core flows¶

5.1 Order ingestion → points allocation¶

flowchart TD
    SRC["Order sources<br/>Swiggy · Zomato · POS · Dashboard"] --> ING["processCommonIngestion<br/>(HTTP / Lambda)"]
    ING --> ORD[("orders — MongoDB")]
    ING --> PQ[("crm_points_allocation_queue<br/>MySQL · status=pending")]
    PQ -->|every 2 min, LIMIT 5000| CRON["crmPointsAllocationQueueCron"]
    CRON --> T{type?}
    T -->|uen_order| F["fraud check → order lookup<br/>→ handleWalletPointsFlow()"]
    T -->|order_reward| OR["handleOrderRewards()"]
    T -->|welcome_point / refer_earn| WP["wallet insert/update"]
    F --> W[("MySQL wallet tables")]
    OR --> W
    WP --> W
    F --> M["handleMilestoneUpgrade()"]
    CRON -->|order not yet in Mongo| RETRY["retry_count++<br/>next_retry_at +15 min<br/>(max 5)"]
    RETRY -.-> PQ

5.2 Campaign / journey → dispatch¶

flowchart TD
    J[("campaign_journeys config")] -->|10 PM daily| PAC["processAutomatedCampaigns"]
    PAC --> BLD["automatedCampaignQueueNew<br/>(audience, template, wallet deduct)"]
    BLD --> SQ[("sms_queue_YYYYMMDD")]
    BLD --> WQ[("whatsapp_queue_YYYYMMDD")]
    BLD --> PN[("push_noti_queue_YYYYMMDD")]
    SQ -->|3 min| SC["smsQueueCron"] --> SMSP["SMS provider"]
    WQ -->|1–3 s| WC["sendWhatsappMessageCron /<br/>sendWhatsappTransactionalCron"] --> GS["Gupshup / Facebook Graph"]
    PN -->|5 min| PC["pushQueueCron"] --> FB["Firebase Admin SDK"]
    GS -.->|delivery webhook ⚠️ no handler cron| WQ

5.3 Loyalty tier mechanics¶

flowchart TD
    W["Customer wallet state<br/>LTO · LTV · lastOrderDate"] --> RFM
    W --> DG
    W --> CI
    subgraph nightly
      RFM["rfmSegments.js — 3 AM<br/>R/F/M over 90 days"] --> SEG["Segments: new · promising · regular ·<br/>champions · need_attention · at_risk · lost"]
      SEG --> RT[("rfm_metrics_tmp · rfm_job_log")]
      DG["loyalty_milestone_downgrade.js — 2 AM<br/>downgrade 1 tier if inactive ≥ downgrade_days"] --> DGO[("customers · addo_contacts_mapping · milestone_logs")]
      CI["customerInsightsCron / segments_count<br/>advanced + channel segment counts"] --> SC[("segment_counts")]
    end
    UP["handleMilestoneUpgrade()<br/>(inline on order points)"] --> DGO

5.4 Analytics / MIS aggregation¶

flowchart TD
    ORD[("orders")] --> AGG["aggregationProcess.js<br/>(IST date math, ETL)"]
    AGG --> A1[("orders_by_day_business")]
    AGG --> A2[("fulfillment_breakdown_by_parent")]
    AGG --> A3[("outlet_metrics_by_day · outlet_toplist_by_day")]
    AGG --> A4[("hour_of_day heatmap · meal_time")]
    AGG --> A5[("top_selling_items · item_revenue · product_pairing")]
    AGG --> A6[("daily_summary_by_parent")]
    SENDS["SMS/WhatsApp/Push sends"] --> CAGG["campaignAggregation.js"] --> CA[("campaign_aggregation")]
    CAGG --> CMIS["campaignMis.js"] --> CM[("campaign_mis")]
    A6 --> DMIS["dashboardMIS.js"] --> DM[("dashboard_mis")]
    DM --> DASH["dashboardControllerNew<br/>(Redis-cached reads)"]

6. Debugging "nothing happened"¶

Symptom	Where to look
Order didn't appear	`crm_queue` for pending (`status:null`/`0`) stuck docs; ingestion cron logs
Points not credited	`crm_points_allocation_queue` `status`/`retry_count`/`error_message`; `handleWalletPointsFlow` logs; check the order actually landed in Mongo (points wait up to 5×15 min for it)
Campaign not delivered	channel queue status (`processed` / `process_status`); business `wallet_balance`; provider errors
Cron seems dead	its `*_cron_status` / `cron_statuses` doc — a prior run stuck at `status:false` blocks the next tick
Segment stale	`rfmSegments` / `customerInsightsCron` last-run status doc

⚠️ Stuck-cron detection is reactive. Crons alert by email only after ~20 minutes of a lock not clearing (e.g. smsQueueCron emails on subtract_time > 20), and only dedupCustomers has a force-close fallback (~4 h). There is no circuit breaker or auto-restart. See Technical Debt.

Cron Catalog · Request Lifecycle · High-Level Architecture · Database · Caching · Modules · Technical Debt