Order Ingestion (Monolith + Prism-Services Edge)¶

Validation legend: ✅ verified in code · 📄 Excel-only · 💻 code-only · ♻️ duplicate/inconsistent · ⚰️ dead code · ⚠️ risk

1. Overview¶

Order Ingestion is the "Ingest & Unify" layer of Prism (see Product Overview §1) — the data heartbeat of the whole platform. Every order from every channel (dine-in POS, the brand's own app/website, uEngage Edge, and POS integrations PetPooja / Rista / Posist / LS Central / OrderPush) lands here first, is buffered in the crm_queue MongoDB collection, then drained by per-POS crons that build the unified Customer 360 / Order data, allocate loyalty points, and trigger digital bills, feedback and journeys.

Why it exists: restaurants receive orders from many disconnected systems. Without a single funnel that normalizes them and unifies the customer by phone number, none of the downstream intelligence (RFM, Campaigns, Loyalty) is possible. crm_queue is the seam of the entire architecture.

The critical architectural fact (⚠️♻️ dual-writer): there are two independent code paths that write orders into the same crm_queue:

Monolith path — Express routes (/injestion/*) → injestionController.js → crm_queue.
Serverless edge — prism-services orders-api Lambda → Kafka (prism-ingest-orders) → orders-consumer Lambda → crm_queue.

Both converge on the same collection with the same numeric source map, but there is no deduplication or shared idempotency key between them (see §9 and High-Level Architecture §9). This is the single most important risk in the module.

2. Business Flow¶

2.1 Product view (the order lifecycle)¶

From Business-Flow §2: every order follows Source → Ingestion API → crm_queue → Processor → Customer 360 → Points/Notifications.

Business rules layered on top (📄 Excel / ✅ where in code): - Cancelled/refunded orders excluded from revenue/AOV/most-selling-item (📄 Dashboard 3.12). ⚠️ Ingestion crons filter some statuses (PetPooja requires Order.status == 1) but there is no universal cancelled-order guard. - Points from cancelled orders clawed back (📄). - Order type split Delivery / Pick-Up / Dine-In, each with its own loyalty earn/burn % — mapped in code as Delivery→1, Takeaway→2, DineIn/Dinning→3. ✅ - Customer identity unified across channels by mobile number (dedup is critical — see Customer-CRM). ✅

2.2 Example journey (a PetPooja dine-in order)¶

A guest pays at a PetPooja terminal. PetPooja POSTs the bill JSON to POST /injestion/petpooja (monolith) or to the Lambda edge POST /crm_api/injestion/petpooja.
Monolith: injestionController.petPooja validates the crm_token against business_configs (falling back to MySQL petPooja_business_mapping), then Queue.create({ data, source:2, insertedAt, status:0 }). Responds {status:1}. ✅
petpoojaOrderIngestion cron (every ~1s) polls {source:2, status:0, "data.data.Order.status":1} in batches of 7000 and calls processLibrary.processPetPoojaData. ✅
That processor upserts the customer (customers) and outlet record (customers_outlets) by (parent_business_id, mobileNo), recomputes LTV/LTO/avg_order_size, inserts the orders doc, and fires loyalty (handleWalletPointsFlow), digital bill, feedback and journey flows.
Queue doc flips to status:1 (done) or status:2 (skipped/errored). ✅

2.3 Edge cases¶

Invalid outlet code / crm_token → rejected with uE_102 Invalid Outlet Code (commonIngestion) or invalid crm token (ls_center/petpooja). ✅
Parent-level order (is_parent == true) via commonIngestion → blocked unless parent_business_id == "6300" (a special-cased tenant → source:5). ✅ ⚠️ hardcoded tenant.
Menu parse failure → uE_103 Error While Parsing Menu. ✅
Order items only written when the parent is is_premium == true. ✅
Dual write → same order can be inserted twice (once by monolith, once by Lambda) with no dedup. ⚠️♻️
Cron stuck → if a cron_stats row stays status:false beyond its threshold, an email alert is sent (hani@uengage.in, prabhpreet@uengage.in). ✅

3. Technical Flow¶

3.1 Full lifecycle (monolith)¶

Source (POS/App/Aggregator)
  → POST /injestion/{pos}            routes/injestion.js  (mounted at root, app.use(injestion_route))
      [middleware varies by POS]     injestion_middleware (hardcoded token) | crm_offer_middleware (authorization) | none
  → injestionController.{handler}    validate token/outlet → Queue.create({data, source:<n>, insertedAt:new Date(), status:0})
  → res.send({status:1})
                                     ── crm_queue (MongoDB) ──
Per-POS cron (node-cron, in web process)
  → poll {source:<n>, status:0} (batch 5k–7k, sort _id:-1)
  → create cron_stats {cron_data, status:false, batch}
  → processLibrary.process<POS>Data(batch, cronId)  |  processCommonIngestion(batch, cronId)
        ├─ parseOrderData()          normalize channel/day/meal counters
        ├─ upsert customers          by (parent_business_id, mobileNo) — LTV+=amt, LTO+=1, avg=LTV/LTO
        ├─ upsert customers_outlets  by (customer_id, parent_business_id, business_id)
        ├─ insert orders             source:"pos", amount, invoiced_at, order_type(1/2/3), code
        ├─ insert order_items        (only if parent is_premium)
        ├─ dashboardMis()            daily MIS rollup
        ├─ handleWalletPointsFlow()  loyalty_type 1/3/null  |  handleMilestoneUpgrade() loyalty_type 2
        ├─ handleOrderRewards(), insertFirstOrderDate(), feedbackFormFlow(), digitalBilFlow(),
        │  sendReferAndEarnMsg(), updateCustomerPreference(), calculateMTOMTVIncrements()
        └─ update crm_queue status → 1 (done) / 2 (skipped)
  → update cron_stats {status:true, end_time}

3.2 Full lifecycle (prism-services edge)¶

POS → httpApi POST /crm_api/injestion/{pos}  (or /crm_api/api/injestion/{pos})
  → orders-api Lambda (src/orders/handlers/api/orders.js)
      produce('prism-ingest-orders', key=Date.now(), value=JSON{pos, payload})   ✅
      return 200 {status:1}
  → Kafka topic prism-ingest-orders (MSK 'flash', batch 10, window 1s, LATEST)
  → orders-consumer Lambda (src/orders/handlers/consumers/orders.js)
      extractMessagesFromEvent() base64-decode
      insert crm_queue {data:payload, source:getSourceId(pos), insertedAt:IST-string, status:0, apiSource:"lambda"}  ✅
      update business_configs {last_uen_order_hit | last_pos_order_hit}          ✅
      (no status update, no dedup, no DLQ) ⚠️

3.3 Source-id map (✅ identical in monolith and Lambda)¶

`source`	POS	Monolith handler	Lambda `getSourceId`	Filter key (Lambda)	Notes
1	uengage (Edge/native)	`crmQueue`	`uengage`	`parent_business_id`+`child_business_id`	updates `last_uen_order_hit`
2	PetPooja	`petPooja`	`petpooja`	`crm_token` (`payload.token`)	validates crm_token
3	LS Central	`ls_center`	`ls_center`	`crm_token`	`injestion_middleware` (hardcoded token)
4	OrderPush / common	`commonIngestion`	`orderPush`	`crm_token` (`payload.outlet_code`)	standard tenants
5	common (tenant 6300)	`commonIngestion`	—	—	💻 monolith-only split for parent 6300
6	Rista	`rista`	`rista`	`crm_token` (`payload.context.branchCode`)	`crm_offer_middleware`
7	Posist	`posist`	`posist`	`crm_token` (`payload.customer_key`)	no route auth ⚠️

⚠️ Correction to map-04: an earlier note claimed 1=Swiggy, 2=Zomato. That is wrong. The authoritative map (verified in both injestionController.js and prism-services/.../consumers/orders.js) is the table above (1=uengage … 7=posist). Swiggy/Zomato arrive inside the uengage/POS payloads and are classified downstream via parseOrderData (swig_count, zom_count).

4. Architecture Diagram¶

4.1 Dual-writer flowchart¶

flowchart TB
    subgraph src["Order Sources"]
        PP[PetPooja]:::pos
        RI[Rista]:::pos
        PO[Posist]:::pos
        LS[LS Central]:::pos
        OP[OrderPush]:::pos
        UE[uEngage Edge / App]:::pos
    end

    subgraph edge["prism-services (Lambda + Kafka)"]
        OAPI[orders-api λ]
        K[[Kafka prism-ingest-orders]]
        OCONS[orders-consumer λ]
    end

    subgraph mono["uengage-crm monolith (Express + node-cron)"]
        RT["/injestion/* routes"]
        IC[injestionController]
        CR{{Per-POS crons<br/>uengage/petpooja/common/…}}
        PL[processLibrary + processCommonIngestion]
    end

    Q[(crm_queue)]:::seam
    DB[(customers / customers_outlets / orders / order_items)]
    CFG[(business_configs)]

    src -->|Lambda path| OAPI --> K --> OCONS -->|insert status:0<br/>apiSource lambda| Q
    OCONS -->|last_*_order_hit| CFG
    src -->|direct path| RT --> IC -->|insert status:0| Q
    CR -->|poll status:0| Q
    CR --> PL --> DB
    PL -->|status 1/2| Q

    classDef pos fill:#eef;
    classDef seam fill:#fe9,stroke:#b80,stroke-width:2px;

4.2 Sequence (edge + monolith converging)¶

sequenceDiagram
    autonumber
    participant SRC as POS
    participant API as orders-api λ
    participant K as Kafka
    participant CN as orders-consumer λ
    participant IC as injestionController (monolith)
    participant Q as crm_queue
    participant CR as ingestion cron
    participant DB as Customer/Order data
    Note over SRC: order fires to EITHER edge OR monolith (sometimes both ⚠️)
    SRC->>API: POST /crm_api/injestion/{pos}
    API->>K: produce base64(order)
    API-->>SRC: 200 {status:1}
    K->>CN: batch ≤10
    CN->>Q: insert {source, status:0, apiSource:lambda}
    SRC-->>IC: POST /injestion/{pos} (direct)
    IC->>Q: insert {source, status:0}
    CR->>Q: poll status:0
    CR->>DB: upsert customer+order, LTV/LTO, points, bill
    CR->>Q: status = 1 / 2

4.3 crm_queue state machine¶

stateDiagram-v2
    [*] --> pending: insert status=0
    pending --> done: processor success (status=1)
    pending --> skipped: validation fail / duplicate (status=2)
    done --> [*]: deleteCrmQueue cron purges status=1
    skipped --> [*]: retained for debug

5. Folder Structure¶

Real files that make up this module.

Monolith (uengage-crm/):

Layer	File	Role	Validation
Route	`routes/injestion.js`	`/injestion/*` + `/api/ingestion/orderPush` (mounted at root)	✅
Route	`routes/crm.js` (lines ~40–42, 387–394)	duplicate `/crm_api/injestion/uengage`+`petpooja`; legacy `/customer/old/injestion`	♻️
Controller	`controllers/injestionController.js`	HTTP gateway: `crmQueue, petPooja, rista, posist, ls_center, commonIngestion, salonOrder, instantProcessing, insertMerchants`	✅
Cron	`controllers/uengageOrderIngestion.js`	drains `source:1` → `processUengageData` (~every 5s)	✅
Cron	`controllers/petpoojaOrderIngestion.js`	drains `source:2` → `processPetPoojaData` (~every 1s)	✅
Cron	`controllers/commonOrderIngestion.js`	drains `source:4` → `processCommonIngestion` (~every 2s)	✅
Cron	`controllers/ristaOrderIngestion.js`	drains `source:6` → `processRistaOrders`	✅
Cron	`controllers/shoutloOrderIngestion.js`	Shoutlo ingestion	✅
Cron	`controllers/posistCron.js`	Posist ingestion (`source:7`)	✅
Helper	`commonFunctions/processCommonIngestion.js`	the core ingest→customer→order→points pipeline	✅
Helper	`commonFunctions/parseOrderData.js`	normalize channel/day/meal counters	✅
Processor	`processLibrary/uengageProcess.js`	`processUengageData`, `processPetPoojaData` (~89KB)	✅
Middleware	`middlewares/injestion_middleware.js`	header `token == "U1CXDZ5S2HQLIE6F"` ⚠️ hardcoded	✅
Middleware	`middlewares/crm_offer_middleware.js`	`authorization` header vs `authorization_token` collection	✅
Model	`models/crm_queue.js`	queue schema (`source, data, insertedAt, status`)	✅
Model	`models/orders_dump.js`	order buffer (collection `order_dump`, `order_id` indexed)	✅
Model	`models/cron_stats.js`	collection `cron_status` (heartbeat/stuck detection)	✅

Serverless (prism-services/):

File	Role
`src/orders/handlers/api/orders.js`	HTTP → Kafka producer (`prism-ingest-orders`)
`src/orders/handlers/consumers/orders.js`	Kafka → `crm_queue` insert + `business_configs` update; `getSourceId`, `getFilterForPos`
`src/services/kafka.js`	`produce()` (hardcoded brokers ⚠️)
`src/services/mongodb.js`	Mongo client singleton (hardcoded Atlas URI ⚠️)
`src/utils/extractMessagesFromEvent.js`	base64 MSK decoder
`serverless-orders.yml`	Lambda + MSK trigger config

6. Database¶

Collection / table	Store	Purpose	Key fields	Written by	Read by
`crm_queue`	Mongo	central ingestion buffer (the seam)	`source` (Num), `data` (Mixed), `insertedAt`, `status`, `apiSource` (lambda)	injestionController, orders-consumer λ	ingestion crons
`cron_status`	Mongo	cron heartbeat / stuck detection	`start_time`, `cron_data`, `status` (bool), `end_time`, `batch`	all ingestion crons	crons, monitoring
`orders`	Mongo	processed order facts	`parent_business_id`, `business_id`, `order_id`, `mobile_number`, `amount`, `invoiced_at`, `order_type`, `source:"pos"`, `code`	processCommonIngestion / processors	RFM, dashboard, order360
`order_items`	Mongo	line items (premium tenants only)	`order_id`, `item_name`, `quantity`, `itemPrice`, `mobile_number`	processors	insights (SKU filters), reports
`customers`	Mongo	Customer 360 master (upserted here)	`(parent_business_id, mobileNo)`, `LTV`, `LTO`, `avg_order_size`, `lastOrderDate`, day/meal/channel counters, `comm_array`	processors	everything downstream
`customers_outlets`	Mongo	per-outlet customer record	`(customer_id, parent_business_id, business_id)`	processors	dedup, insights
`business_configs`	Mongo	tenant config; last-hit timestamps	`parent_business_id`, `child_business_id`, `crm_token`, `is_parent`, `is_premium`, `loyalty_type`, `last_uen_order_hit`, `last_pos_order_hit`	orders-consumer λ, controllers	validation, crons
`order_dump`	Mongo	legacy raw order buffer	`order_id` (indexed), `mobile_number`, `amount`, `source`	(legacy)	reporting
`common_ingestion_unprocessed_queue`	Mongo	validation-failure records	error payload	processCommonIngestion	debug
`petPooja_business_mapping`	MySQL	PetPooja crm_token fallback	`businessId`, `petpooja_outlet_id`	integration setup	`petPooja` handler
`crm_points_allocation_queue`	MySQL	loyalty point allocation queue	`status:'pending'`, type `uen_order`/`order_reward`	processors (via points flow)	`crmPointsAllocationQueueCron`

Keys/relationships: crm_queue.source selects the processor; mobileNo unifies the customer; parent_business_id scopes tenancy; crm_token/outlet_code/branchCode/customer_key map a POS payload → business (see §3.3).

Indexes: ⚠️ crm_queue has no indexes (model defines none). Dequeue queries {source, status} scan; a compound (source, status, _id) index is the top recommendation (map-03 §5.1). order_dump.order_id is indexed. customers uses a compound (parent_business_id, mobileNo) (index hint referenced in the injestion cron).

Common queries:

// dequeue one POS's pending batch
db.crm_queue.find({ source: 2, status: 0, "data.data.Order.status": 1 }).sort({ _id: -1 }).limit(7000)
// customer upsert (dedup by mobile)
db.customers.updateOne({ parent_business_id, mobileNo }, { $inc:{LTV:amt,LTO:1}, $set:{lastOrderDate} }, { upsert:true })

7. APIs¶

Monolith ingestion routes¶

Mounted at root (index.js: app.use(injestion_route)), method POST, response {status:1, message:"data inserted successfully"} on success / {status:false, message:"something went wrong"} on error.

Route	Handler	`source`	Auth middleware	Validation	Failure cases
`/injestion/uengage`	`crmQueue`	1	none	body `parentBusinessId`,`businessId`	generic 500
`/injestion/petpooja`	`petPooja`	2	none (body crm_token)	crm_token ∈ `business_configs` (MySQL fallback)	`invalid crm token`
`/injestion/rista`	`rista`	6	`crm_offer_middleware`	`authorization` header	`uE_101 Authentication Failed`
`/injestion/posist`	`posist`	7	none ⚠️	—	posist bill capture msg
`/injestion/ls_center`	`ls_center`	3	`injestion_middleware` (hardcoded token)	crm_token non-empty ∈ configs	`token in required` / `invalid token`
`/api/ingestion/orderPush`	`commonIngestion`	4/5	`crm_offer_middleware`	`outlet_code` ∈ configs; block `is_parent` unless 6300	`uE_102 Invalid Outlet Code`, `uE_103 Error While Parsing Menu`
`/injestion/salon/order`	`salonOrder`	(async)	none	—	processes immediately (no queue)
`/injestion/instant/order`	`instantProcessing`	(async)	none	—	immediate processing

♻️ The same /injestion/uengage and /injestion/petpooja are also registered under /crm_api in routes/crm.js (lines 41–42) — two mount points for the same handlers.

Example payload (uengage, POST /injestion/uengage) — verify against live POS contract:

{ "parentBusinessId": "4580", "businessId": "9001",
  "order_id": "UE-88231", "mobile_no": "9876543210",
  "amount": 540, "orderType": "Delivery", "order_from": "Swiggy",
  "orderDate": "2026-07-03 20:14:00", "items": [{ "name":"Paneer Tikka","qty":1,"price":320 }] }

Serverless edge routes (`prism-services`)¶

Route	Lambda	Behaviour
`POST /crm_api/injestion/{pos}`	`orders-api`	produce to Kafka → 200 `{status:1}`
`POST /crm_api/api/injestion/{pos}`	`orders-api`	same

Consumer inserts {data, source, insertedAt(IST string), status:0, apiSource:"lambda"}. No auth, no validation, no DLQ, no dedup. ⚠️

8. Code Walkthrough¶

injestionController.js — one handler per POS. Each is a thin adapter: validate the tenant token, build {data:req.body, source:<n>, insertedAt:new Date(), status:0}, Queue.create(...), respond. commonIngestion is heavier: it validates outlet_code, enforces the is_parent/6300 rule, and picks source:4 vs 5. salonOrder/instantProcessing bypass the queue and process synchronously.

commonOrderIngestion.js / petpoojaOrderIngestion.js / uengageOrderIngestion.js — all follow the same shape: cron.schedule → check last cron_stats for stuck (>60 min → email alert) → poll {source, status:0} (batch 5k–7k) → create cron_stats → call processor → mark cron_stats done.

commonFunctions/processCommonIngestion.js — the workhorse. Per queue doc: validate outlet & payload → upsert customers (dedup by (parent_business_id, mobileNo), upsert:true, catches dup-key 11000 and refetches) → upsert customers_outlets → insert orders (map orderType → 1/2/3) → insert order_items if premium → dashboardMis() → loyalty (handleWalletPointsFlow for loyalty_type 1/3/null; handleMilestoneUpgrade for 2) → handleOrderRewards, insertFirstOrderDate, feedbackFormFlow, digitalBilFlow, sendReferAndEarnMsg, calculateMTOMTVIncrements → set crm_queue.status = 1 | 2. LTV/LTO/AOV are recomputed here: LTV += amount; LTO += 1; avg_order_size = LTV/LTO.

commonFunctions/parseOrderData.js — parseData(data_source, data) derives mutually-exclusive counters from data.date and data.order_from: pos_count / zom_count / swig_count, meal (lunch 11–16, dinner 19–24, mid_night 0–6), and day-of-week counters. These feed customer preference fields used by RFM day/time filters.

prism-services consumer — getSourceId(pos) (the authoritative map), getFilterForPos(payload, pos) (per-POS key extraction), insertOrderIntoMongoDB (insert + business_configs last-hit update). No status transitions — the monolith crons still own draining.

Call hierarchy: route → injestionController.<pos> → crm_queue … (async) … cron.schedule → poll → processLibrary.process<POS>Data / processCommonIngestion → {customers, customers_outlets, orders, order_items, points, notifications} → crm_queue.status.

9. Business Rules¶

Rule	Detail	Validation
Dual-writer to crm_queue	Monolith and Lambda both insert; identical `source` map; no dedup / no idempotency key	♻️⚠️
Source-id map	1=uengage,2=petpooja,3=ls_center,4=orderPush,5=common(6300),6=rista,7=posist	✅ (map-04 "1=Swiggy" is wrong)
Dedup by mobile	`customers` upsert keyed on `(parent_business_id, mobileNo)`; dup-key retry	✅
Order-type mapping	Delivery→1, Takeaway→2, DineIn/Dinning→3	✅
Order items = premium only	`order_items` inserted only when parent `is_premium`	✅
Tenant 6300 special case	commonIngestion routes 6300 to `source:5`, allows parent-level orders	💻⚠️ hardcoded
Cancelled orders	Excel: exclude from revenue/points; code has no universal cancelled filter (PetPooja checks `Order.status==1`)	📄 spec, ⚠️ partial in code
status semantics	0=pending, 1=done, 2=skipped/errored	✅
insertedAt inconsistency	monolith writes a JS `Date`; Lambda writes an IST string	♻️
Hardcoded ingestion token	`injestion_middleware` compares to `"U1CXDZ5S2HQLIE6F"`	⚠️ security
Posist route unauthenticated	`/injestion/posist` has no route-level auth	⚠️
Sync bypass	`salonOrder`/`instantProcessing` skip the queue entirely	✅
queue cleanup	`deleteCrmQueue` purges `status:1` and stale source-6 non-terminal events (every 2 days)	✅ (cf. recent commit "remove duplicate source-6 order entries")

10. Performance¶

Buffering decouples spikes: the Lambda edge + Kafka absorb bursty POS traffic; the monolith drains at its own pace via node-cron. ✅
Large batches: crons pull 5k–7k docs per tick, sort:{_id:-1} (newest first). ✅
Polling architecture ⚠️: no message broker on the monolith side — crons poll crm_queue on tight schedules (~1–5s). Wasted queries when idle; potential thundering-herd under load (map-04 §5.3).
Missing indexes ⚠️: crm_queue has none — dequeue {source,status} scans grow with backlog. Add (source, status, _id).
In-process crons ⚠️: all ingestion crons run inside the web process → ingestion load couples with API latency (HLA §9).
No DLQ / retry on Lambda consumer ⚠️: a bad message can crash the consumer; offsets are LATEST so historical messages are skipped.
Cleanup: deleteCrmQueue (every 2 days) keeps the collection bounded — but there is no TTL index (map-03 §5.4).

11. Logging¶

Monolith crons: heavy console.log plus cron_status rows (start_time, end_time, batch) as the durable trace. Stuck detection emails carry the offending cron_stats._id.
Lambda: structured console.log in CloudWatch — notably a lag metric per message: {createdAt: timestamp, processedAt: now, lag: (now-timestamp)/1000} (Kafka → consume latency). Errors console.error("Error inserting payload into MongoDB", err).
Validation failures: persisted to common_ingestion_unprocessed_queue for replay/debug.
Trace correlation: cron_stats._id links a batch run to its log lines; apiSource:"lambda" on a crm_queue doc identifies the writer path (edge vs monolith) — the primary way to attribute a duplicate.

12. Monitoring¶

Signal	Where	Healthy
Queue backlog	`crm_queue.countDocuments({status:0})` per source	Small & draining; sustained growth ⇒ cron down
Cron liveness	`cron_status` newest row per `cron_data`	`status:true` with recent `end_time`; no row `status:false` > threshold
Stuck alert	email to hani@/prabhpreet@	none firing
Ingest freshness	`business_configs.last_uen_order_hit` / `last_pos_order_hit`	within minutes of live orders
Lambda lag	CloudWatch `lag` metric	low seconds
Skipped rate	`crm_queue.countDocuments({status:2})`	low; spikes ⇒ validation/config issue
Duplicate rate	orders with same `order_id`	~0 (dual-writer risk)

"Healthy" = per-source status:0 backlog near zero, every cron_data heartbeat fresh, last_*_order_hit current, Lambda lag in single-digit seconds.

13. Troubleshooting¶

Symptom	Root cause	Resolution / commands
Orders not appearing in Customer 360	cron stuck (row `status:false`) or `crm_queue` backlog	`db.cron_status.find({cron_data:"petpooja"}).sort({_id:-1}).limit(1)`; restart PM2 process; check backlog `db.crm_queue.countDocuments({source:2,status:0})`
Duplicate orders / inflated LTO	dual-writer (both edge + monolith ingested)	Group by `order_id`; check `apiSource`; dedup + patch LTV/LTO; long-term add idempotency key
POS returns "invalid crm token"	`crm_token`/`outlet_code` not in `business_configs`	Verify tenant onboarding; check MySQL `petPooja_business_mapping` fallback
commonIngestion `uE_102`	outlet_code missing/unknown, or parent-level order (not 6300)	Confirm outlet mapping; parent orders only allowed for 6300
Queue growing, cron "running" forever	stuck cron never marked done	Wait for stuck-email / force restart; inspect processor exception logs
Lambda consumer crashlooping	bad message, Mongo error (no DLQ)	CloudWatch logs; fix/skip poison message; consider adding DLQ
Points not allocated	`crm_points_allocation_queue` not draining	Check `crmPointsAllocationQueueCron`; loyalty_type routing (1/3/null vs 2)

14. FAQs¶

Why are there two ways to ingest the same order? Historical migration: prism-services (Lambda+Kafka) was added to absorb spiky POS load; the monolith routes predate it and still run. Both write crm_queue. Reconciling them is a roadmap item.
Is source:1 Swiggy? No — source:1 is uengage. Swiggy/Zomato are classified from the payload's order_from into swig_count/zom_count by parseOrderData.
Who marks a queue doc processed? Only the monolith crons (status:1/2). The Lambda consumer never updates status.
Where does dedup happen? At the customer level (customers upsert by mobile), not at the queue level — so duplicate queue rows can still inflate orders.
Why do PetPooja orders filter Order.status==1? Only completed/valid PetPooja bills are ingested; other statuses are ignored at poll time.
Difference between orders and order_dump? orders is the live processed collection; order_dump (order_id indexed) is a legacy raw buffer.

15. Cheat Sheet¶

SEAM:        crm_queue {source, data, insertedAt, status(0=pending,1=done,2=skip), apiSource}
SOURCE MAP:  1 uengage | 2 petpooja | 3 ls_center | 4 orderPush | 5 common(6300) | 6 rista | 7 posist
             (map-04 "1=Swiggy" is WRONG)
WRITERS:     monolith injestionController  +  prism-services orders-consumer λ  → SAME collection ⚠️ no dedup
ROUTES (POST, root mount):
  /injestion/uengage(crmQueue,src1,noauth)  /injestion/petpooja(src2,token)  /injestion/rista(src6,crm_offer)
  /injestion/posist(src7,NOAUTH⚠️)  /injestion/ls_center(src3,hardcoded-token)  /api/ingestion/orderPush(src4/5,crm_offer)
EDGE:        POST /crm_api/injestion/{pos} → orders-api λ → Kafka prism-ingest-orders → orders-consumer λ
CRONS:       uengageOrderIngestion(src1) petpoojaOrderIngestion(src2) commonOrderIngestion(src4) ristaOrderIngestion(src6) posistCron(src7)
             poll {source,status:0} batch 5k-7k → process<POS>Data / processCommonIngestion → status 1/2
PIPELINE:    upsert customers(dedup mobile) → customers_outlets → orders → order_items(premium) → points → bill/feedback/journey
MODELS:      models/crm_queue.js  models/cron_stats.js(→cron_status)  models/orders_dump.js
GOTCHAS:     dual-writer duplicates ⚠️ | cancelled not universally excluded ⚠️ | hardcoded token U1CXDZ5S2HQLIE6F ⚠️
             insertedAt Date(monolith) vs IST-string(lambda) ♻️ | no crm_queue index | tenant 6300 special-cased

Downstream: Customer-CRM (customers built here), Customer-Intelligence-RFM (consumes orders), Loyalty (points from ingestion), Dashboard-Analytics, Digital-Bills, Feedback-NPS.
Cross-cutting: High-Level Architecture · Queues · Database · API · Security · Business-Flow §2.

Knowledge Tests¶

Level 1 — MCQs¶

In crm_queue, what does source:6 mean? a) Zomato b) uengage c) Rista d) Posist — c ✅.
Which two paths write to crm_queue? a) only crons b) injestionController (monolith) + orders-consumer Lambda c) only Kafka d) only Redis — b.
crm_queue.status values are: a) pending/sent/failed b) 0 pending / 1 done / 2 skipped c) true/false d) 0/1 only — b.
Where does customer dedup happen? a) in crm_queue b) in Kafka c) on the customers upsert by (parent_business_id, mobileNo) d) in Redis — c.
order_items are written only when… a) always b) parent is_premium c) COD orders d) dine-in only — b.

Level 2 — Scenarios¶

A brand reports a customer's LTO is double what they expect. All their orders come from PetPooja. What single architectural feature likely explains it, and how would you confirm and remediate? (dual-writer; group by order_id / check apiSource.)
During a lunch rush, orders stop appearing in dashboards for one tenant only. crm_status for petpooja shows status:false from 45 minutes ago. Walk through diagnosis and recovery.

Level 3 — Code reading¶

Open commonFunctions/processCommonIngestion.js. Identify exactly where LTV, LTO, and avg_order_size are computed, and explain how a MongoDB duplicate-key error (11000) is handled during the customer upsert.

Level 4 — Architecture¶

The dual-writer to crm_queue risks duplicate orders. Propose an idempotency design that works across both the monolith and the Lambda edge without a shared DB transaction. Address the natural key, where to enforce it, and backfill. Reference High-Level Architecture §9.

Level 5 — Production debugging¶

The orders-consumer Lambda is crashlooping and Kafka consumer lag is climbing. crm_queue inserts from apiSource:"lambda" have stopped, but monolith inserts continue. Diagnose (poison message + no DLQ + errors rethrown crash the batch, offsets not committed), give immediate mitigation and the structural fix.

Practical Assignments¶

Trace an order end-to-end: POST a synthetic order to /injestion/uengage on a dev tenant, watch it land in crm_queue (status:0), observe the uengageOrderIngestion cron drain it, and confirm the resulting customers/orders docs and cron_status heartbeat. Document each hop with timestamps.
Reproduce the dual-writer duplicate: send the same order to both /injestion/petpooja (monolith) and /crm_api/injestion/petpooja (edge). Show two crm_queue docs (one apiSource:"lambda") and the resulting duplicate effect; propose the idempotency key.
Add a new POS source: wire a hypothetical source:8 (route + handler + getSourceId in the Lambda + a draining cron). Enumerate every file you must touch — a concrete map of the module's surface area.