OMS — Omni-Channel Order Management System

A production-grade, fully async, multi-tenant Order Management System built with Python 3.12, FastAPI, and a polyglot persistence layer — PostgreSQL, MongoDB, Redis, and Elasticsearch. Designed for high-throughput omni-channel retail with intelligent multi-node sourcing, real-time inventory reservation, a fully automated fulfillment pipeline, and complete data isolation between organizations and environments.

1 Architecture Overview

AI-Native 5-Layer Architecture

Request Lifecycle

1. Client POSTs an order to POST /orders
2. FastAPI validates the request via Pydantic v2 schemas
3. Order is written to PostgreSQL and indexed in Elasticsearch
4. Order-created event is written to MongoDB audit log
5. A source_order task is enqueued on the Redis-backed sourcing Celery queue
6. Sourcing Engine evaluates active rules — may route via A/B experiment or AI_ADAPTIVE strategy
7. Pipeline tasks flow: sourcing → fulfillment (pick → pack) → carrier (label + ship) → notifications → webhooks
8. Delivery outcome is labeled by the learning worker and fed back into the pattern store

Directory Structure

D:\OMS\
├── Dockerfile                      # Multi-stage Python 3.12 image
├── docker-compose.yml              # All 9 services
├── requirements.txt                # All Python dependencies
├── .env                            # Local environment variables
│
├── app/
│   ├── main.py                     # FastAPI app, lifespan, router registration
│   ├── config.py                   # Pydantic Settings (reads .env)
│   │
│   ├── database/
│   │   ├── postgres.py             # Async SQLAlchemy engine + session factory
│   │   ├── mongodb.py              # Motor async client + index creation
│   │   ├── redis_client.py         # aioredis pool + cache helpers
│   │   └── elasticsearch_client.py # Async ES client + index mapping creation
│   │
│   ├── models/postgres/
│   │   ├── order_models.py         # Order, OrderItem, FulfillmentAllocation, Shipment
│   │   ├── inventory_models.py     # InventoryItem, InventoryAdjustment, InventoryReservation
│   │   ├── node_models.py          # FulfillmentNode + NodeType/NodeStatus enums
│   │   └── sourcing_rule_models.py # SourcingRule + SourcingStrategy/ConditionOperator enums
│   │
│   ├── schemas/                    # Pydantic v2 request/response schemas
│   ├── routers/                    # FastAPI route handlers (7 files, 44 endpoints)
│   ├── services/
│   │   ├── sourcing_engine.py      # Intelligence core (rule eval, node scoring, allocation)
│   │   └── webhook.py              # HMAC-SHA256 signed delivery
│   │
│   ├── models/postgres/
│   │   ├── ...existing models...
│   │   └── ai_models.py            # AIProposal, AIExperiment, CustomAttributeDefinition,
│   │                               #   UIWidget, SourcingOutcomeLabel + enums
│   │
│   ├── services/
│   │   ├── sourcing_engine.py      # Core + AI_ADAPTIVE + experiment traffic splitting
│   │   ├── ai_sourcing.py          # AISourcingAdvisor — KubeAI node scoring
│   │   ├── pattern_discovery.py    # PatternDiscoveryService — nightly cluster aggregation
│   │   └── schema_evolution.py     # SchemaEvolutionEngine — safe additive schema changes
│   │
│   └── workers/
│       ├── celery_app.py           # Celery factory + 7 queues + beat schedule
│       ├── sourcing.py             # source_order task (writes sourcing_outcomes to MongoDB)
│       ├── fulfillment.py          # start_picking, complete_packing, reset counters
│       ├── carrier.py              # book_shipment, simulate_delivery, sync tracking
│       ├── notifications.py        # Email/SMS notification tasks
│       ├── webhooks.py             # dispatch_webhook, retry_failed_webhooks
│       ├── connectors.py           # sync_fulfillment_to_connector, poll_amazon_orders
│       └── learning.py             # label_sourcing_outcomes, discover_patterns,
│                                   #   update_node_performance, evaluate_ai_experiments
│
├── frontend/                       # React + Vite + TypeScript GUI (port 3000)
├── scripts/
│   └── seed.py                     # Seeds all 4 databases with realistic data
└── tests/
    └── test_imports.py             # 13 import + unit tests

2 Tech Stack

3 PostgreSQL Models

fulfillment_nodes

orders

inventory_items

Per-node, per-SKU stock levels with three counters:

• quantity_on_hand — physical stock in the warehouse
• quantity_reserved — soft-reserved by active allocations
• quantity_available = on_hand − reserved (what sourcing can use)

fulfillment_allocations

Bridges an order item to a specific fulfillment node. One order can have multiple allocations (split fulfillment). Tracks the full picking → packing → shipping timeline with timestamps for each transition.

sourcing_rules

Configurable rules evaluated in priority order (ascending). Each rule has:

• conditions — JSON array of {field, operator, value} tuples
• strategy — DISTANCE_OPTIMAL, COST_OPTIMAL, STORE_NEAREST, INVENTORY_RESERVATION, LEAST_COST_SPLIT, AI_ADAPTIVE, or AI_HYBRID
• allowed_node_types, required_capabilities — node pre-filter
• max_split_nodes, cost_weight, distance_weight — algorithm parameters

Control-Plane Models (org_models.py)

These four tables live exclusively in oms_db — the shared control-plane database — and are never replicated into tenant databases.

environments table — key columns

AI Models (ai_models.py)

Other Models

4 MongoDB Collections

Database: oms_events — transactional / operational

Latency-sensitive collections written on the hot request path. Fast point lookups by order_id.

Database: oms_ai_learning — analytical / AI pipeline

High-volume write collections used by the AI learning loop. Aggregation-heavy reads run here without contending with order audit lookups. TTLs bound storage growth automatically.

5 Redis Key Schema

Redis is used as the Celery broker (DB 1) and result backend (DB 2), keeping pipeline task state separate from application cache (DB 0). The env:{id} cache key is the critical performance optimization for the multi-tenant middleware — without it, every API request would require a synchronous lookup into oms_db to resolve the active environment.

6 Elasticsearch Indexes

oms_orders

Optimized for order search. Documents are indexed on order creation and updated on every status transition.

oms_products

Product catalog search backed by MongoDB documents synced to Elasticsearch.

7 Orders API /orders

Create Order — Request Body

{
  "channel": "WEB",
  "fulfillment_type": "SHIP_TO_HOME",
  "customer_email": "alice@example.com",
  "customer_name": "Alice Smith",
  "line_items": [
    {
      "sku": "SKU-WIDGET-A",
      "product_name": "Premium Widget A",
      "quantity": 2,
      "unit_price": 29.99
    }
  ],
  "shipping_address": {
    "address1": "123 Main St",
    "city": "New York",
    "state": "NY",
    "postal_code": "10001",
    "latitude": 40.7484,
    "longitude": -73.9967
  }
}

Order Status Values

8 Inventory API /inventory

Adjustment Reason Codes

Adjust Stock Example

{
  "quantity_delta": 50,
  "reason": "RECEIVED",
  "notes": "PO-2024-0042 inbound shipment",
  "created_by": "ops_user"
}

9 Fulfillment Nodes API /nodes

10 Sourcing Rules API /sourcing-rules

11 Search API /search

Supports: fuzzy matching, multi-field search, date/amount range filters, pagination, sort order. Results include a score field for relevance ranking.

Search Request Example

{
  "query": "John Doe",
  "status": "DELIVERED",
  "from_date": "2024-01-01",
  "to_date": "2024-12-31",
  "min_amount": 50,
  "page": 1,
  "page_size": 20,
  "sort_by": "created_at",
  "sort_order": "desc"
}

12 Analytics API /analytics

Dashboard Response Shape

{
  "period_start": "2024-01-01",
  "period_end": "2024-12-31",
  "total_orders": 1247,
  "total_revenue": 184293.50,
  "avg_order_value": 147.79,
  "orders_by_status": { "DELIVERED": 891, "SHIPPED": 124, ... },
  "orders_by_channel": [
    { "channel": "WEB", "count": 623, "total_revenue": 98241.00, "percentage": 49.9 }
  ],
  "top_nodes": [
    { "node_id": "...", "node_name": "DC-EAST", "total_allocations": 412, "capacity_utilization": 0.61 }
  ],
  "inventory_alerts": [
    { "sku": "SKU-GADGET-X", "node": "STR-MIA-01", "available": 2, "reorder_point": 10 }
  ]
}

13 Webhooks API /webhooks

14 Connectors API /connectors

Superadmin-only CRUD for managing platform integrations. The inbound webhook receiver is public (authenticated via HMAC) — all other endpoints require a superadmin JWT.

Create Connector Example

curl -X POST http://localhost:8001/connectors/ \
  -H "Authorization: Bearer {superadmin_token}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "My Shopify Store",
    "connector_type": "SHOPIFY",
    "direction": "BIDIRECTIONAL",
    "config": {
      "shop_url": "mystore.myshopify.com",
      "access_token": "shpat_xxxxxxxxxxxx",
      "webhook_secret": "my-hmac-secret",
      "api_version": "2024-01"
    }
  }'

Note: Sensitive config keys (access_token, webhook_secret, api_key, etc.) are always masked as *** in all API responses.

15 AI Architect /architect

The AI Architect is a superadmin-only control center that closes the loop between order data and sourcing intelligence. It is the place where the system observes its own performance, proposes improvements, and waits for a human to say yes before changing anything.

What does the AI Architect do?

Normally, sourcing rules are manually written by admins — you define conditions, pick a strategy, and the engine applies it. The AI Architect adds a self-improving layer on top:

1. It watches every sourcing decision — every order allocation writes a record to MongoDB (sourcing_outcomes) capturing which node was chosen, at what cost, via which strategy.
2. It labels outcomes once the order is delivered — after delivery, the learning worker scores the decision: was it fast? was the cost accurate? did the item backorder or get returned?
3. It discovers patterns nightly — orders are grouped by channel, region, order-size bucket, and fulfillment type. Within each group, it ranks nodes by their historical outcome scores and compares AI-sourced orders against rule-based ones.
4. It proposes improvements — when a pattern has enough data and AI consistently outperforms the baseline by ≥10%, the system drafts a new sourcing rule proposal and queues it for your review.
5. You approve or reject it — nothing changes in the system until you click Approve → Apply. Every action is reversible.

Tab 1 — Proposals

Proposals are the core output of the learning loop. They represent changes the AI wants to make, held in a queue until a human reviews them.

Where do proposals come from?

• Pattern discovery (nightly) — when the discover_patterns Celery task finds a cluster where AI_ADAPTIVE achieves ≥10% better outcome scores than DISTANCE_OPTIMAL across ≥50 labeled orders, it auto-creates a pending sourcing rule proposal.
• Experiment completion (daily) — when an A/B experiment accumulates ≥50 labeled orders per arm and declares a winner, it creates a proposal to promote the winning strategy as a permanent rule.

Proposal lifecycle — step by step

Reading a proposal

Each proposal card shows:

• Title — plain-language summary, e.g. "Use AI_ADAPTIVE for WEB|NY|100-250|SHIP_TO_HOME cluster (+14.2% better outcomes)"
• Confidence score — 0–1 bar combining improvement magnitude and sample size. A score of 0.8+ means strong evidence; below 0.5 means the data is suggestive but not conclusive.
• Rationale — the numbers behind the proposal: how many orders, what average outcome scores, which thresholds were met.
• Proposal data — the exact JSON that will be written to the database if you click Apply. For a sourcing rule proposal this includes: name, strategy, priority, and the auto-generated conditions derived from the cluster's dimensions.
• Status badge — PENDING (yellow), APPROVED (blue), APPLIED (green), REJECTED (red), ROLLED_BACK (gray).

Filtering proposals

Use the status filter tabs (All / Pending / Approved / Applied / Rejected) at the top of the Proposals tab to focus on what needs action. "Pending" is the most important queue — these are waiting for your decision.

Tab 2 — Patterns

The Patterns tab shows you what the system has learned from historical orders, grouped by order characteristics. This is the raw intelligence that proposals are built from.

What is a pattern cluster?

A cluster is a combination of four dimensions that describes a type of order:

Example cluster key: WEB|NY|100-250|SHIP_TO_HOME — web orders shipped to New York customers, order value $100–$250.

Reading a pattern card

• Sample count — total number of labeled (delivered) orders in this cluster. More samples = more reliable data. Below 50 samples, treat it as indicative only.
• Node performance table — nodes ranked by avg_outcome_score for this cluster. Shows average delivery hours, average cost, and how many orders they've handled.
• Best node — highlighted at the top. This is the node the system would prefer for AI_ADAPTIVE sourcing within this cluster.

Patterns are updated nightly. If a cluster has fewer than 10 samples, the AI sourcing advisor will not use it (fallback to DISTANCE_OPTIMAL instead).

What to look for

• Clusters with high sample counts (100+) and a clear top node are good candidates for a dedicated sourcing rule.
• If the top node's avg_delivery_hours is significantly lower than others, AI_ADAPTIVE will reliably pick it.
• Clusters where multiple nodes have similar scores indicate the order type is not sensitive to node choice — no specific rule needed.

Tab 3 — Experiments

Experiments let you A/B test two sourcing strategies on live traffic before committing to one. A percentage of orders are routed to the "treatment" strategy; the rest continue with the "control". The system measures outcomes on both arms and declares a winner when the data is conclusive.

Creating an experiment

Click New Experiment and fill in:

How traffic splitting works

When an order is sourced, the engine checks all running experiments whose filter conditions match the order. For the first matching experiment, it uses a random draw:

# In sourcing_engine._check_experiment():
if random.random() * 100 < exp.traffic_split_pct:
    strategy = exp.strategy_b   # treatment arm (e.g. AI_ADAPTIVE)
else:
    strategy = exp.strategy_a   # control arm (e.g. DISTANCE_OPTIMAL)

The experiment ID is stamped on the sourcing_outcomes MongoDB document so the learning worker can attribute the outcome to the correct arm.

Reading experiment results

The results panel shows per-arm data once labeled outcomes are available (orders must reach DELIVERED status first):

• Total orders — how many orders were routed to each arm (labeled + unlabeled).
• Labeled orders — orders that have reached DELIVERED and have an outcome_score computed.
• Avg outcome score — the primary metric. Higher is better. A difference of 0.05 or more (on a 0–1 scale) triggers automatic winner declaration.
• Avg delivery hours — secondary indicator. Lower is better.
• Backorder rate % — what % of orders in this arm had an item backordered.

When does an experiment end?

The evaluate_ai_experiments Celery task runs daily at 03:00 UTC and checks every running experiment. It declares a winner when all three conditions are met:

• Each arm has ≥ 50 labeled outcomes
• The absolute score difference between arms is ≥ 0.05
• (If the difference is smaller, the experiment continues collecting data)

When a winner is declared, the experiment status changes to COMPLETED and a new sourcing_experiment proposal is created, suggesting you promote the winning strategy as a permanent sourcing rule.

Managing experiments

• Pause — stops new orders from being routed to this experiment. Existing labeled outcomes are preserved. Useful if you see an unexpected issue with the treatment arm.
• Resume — re-activates a paused experiment from where it left off.

Tab 4 — Node Performance

Shows rolling performance metrics for every fulfillment node, computed by the update_node_performance task every 4 hours.

• 7-day view — reflects recent changes in node behavior. Use this to spot a node that has degraded in the last week (rising backorder rate, slower delivery).
• 30-day view — a more stable baseline. Use this to compare nodes over a meaningful time horizon.

Each node card shows: orders fulfilled, avg outcome score, avg delivery hours, avg cost, backorder rate %, and return rate %. Nodes are ranked by avg outcome score descending. A node consistently in the top 3 by outcome score is a strong candidate to be the "best node" for matching order clusters.

Setting up AI sourcing on your orders

Follow these steps to enable AI-powered sourcing for a subset of your orders:

Option A — Direct rule (if you already trust the data)

1. Go to Sourcing Rules in the sidebar.
2. Create a new rule. Set Strategy to AI_ADAPTIVE.
3. Add conditions to restrict which orders this rule applies to — e.g. channel = WEB and shipping_state = CA.
4. Set a priority number lower than your other rules (lower number = higher priority, e.g. 10).
5. Toggle the rule Active.
6. From this point, matching orders will be scored by KubeAI using historical pattern data. If pattern data is insufficient, it automatically falls back to DISTANCE_OPTIMAL.

Option B — A/B experiment first (recommended for production)

1. Go to AI Architect → Experiments tab.
2. Create an experiment: Strategy A = DISTANCE_OPTIMAL, Strategy B = AI_ADAPTIVE, traffic split = 10%.
3. Add filter conditions to target a specific order segment (e.g. a single channel or region).
4. Wait until the experiment accumulates ≥50 labeled orders per arm (orders must be delivered). This typically takes 1–4 weeks depending on volume.
5. When the experiment completes, review the auto-generated proposal in the Proposals tab.
6. If the results are good, approve and apply the proposal. The new sourcing rule will be inserted (inactive); activate it in Sourcing Rules to go live.

Option C — Wait for automatic proposals

If you use AI_ADAPTIVE on some orders (via a rule), the learning pipeline will observe outcomes over time. Nightly pattern discovery will automatically create proposals when a specific order cluster shows ≥10% improvement. You just need to review the Proposals tab periodically.

Common workflows

"I want to know if AI sourcing is working better than my current setup"

1. Go to AI Architect → Experiments, create an experiment comparing your current rule strategy against AI_ADAPTIVE.
2. Monitor the Node Performance tab to ensure AI-sourced orders are going to high-performing nodes.
3. After sufficient data is collected, check the experiment results. Compare avg_outcome_score, avg_delivery_hours, and backorder_rate between arms.
4. If AI_ADAPTIVE wins, approve and apply the auto-generated proposal.

"I received a proposal — should I approve it?"

Look at three things:

• Confidence score: Above 0.7 — generally safe to approve. Below 0.5 — let more data accumulate first (reject and wait).
• Sample count in rationale: More than 100 total samples with 20+ AI samples is a solid basis. If the AI arm had only 10 orders, the improvement could be noise.
• Improvement %: A 10–15% improvement is meaningful. 50%+ suggests something unusual — verify that the cluster key is sensible and not covering a very small order type.

When you click Apply, the sourcing rule is created inactive. Review it in Sourcing Rules before activating it to see the exact conditions and confirm they look right.

"I applied a proposal but the new sourcing rule isn't performing as expected"

1. Go to AI Architect → Proposals, find the applied proposal, and click Rollback. This deletes the inserted sourcing rule.
2. The original routing behavior resumes immediately (no restart needed).
3. Alternatively, go to Sourcing Rules and deactivate the rule rather than rolling back — this preserves it for future reference.

"How do I see what node was chosen for a specific order and why?"

Go to Orders → [order] → Sourcing Decision. The decision trail shows: strategy used, rule matched, every candidate node with its distance/cost/score, and — for AI-sourced orders — the AI score and KubeAI's one-sentence reasoning per node.

API Reference

All endpoints require superadmin JWT. Base path: /architect.

Proposals

Patterns & Performance

Experiments

16 Sourcing Rules Engine

File: app/services/sourcing_engine.py

The engine is the intelligence core. It runs every time an order needs to be sourced, evaluating prioritized rules to determine which fulfillment nodes should handle each SKU.

Processing Pipeline

Seven Sourcing Strategies

Condition Operators

Haversine Distance Formula

Used for all distance-based sourcing calculations. Accuracy: ±0.5% vs actual road distance.

def haversine_km(lat1, lon1, lat2, lon2):
    R = 6371.0  # Earth radius in km
    φ1, φ2 = radians(lat1), radians(lat2)
    Δφ = radians(lat2 - lat1)
    Δλ = radians(lon2 - lon1)
    a = sin(Δφ/2)**2 + cos(φ1)*cos(φ2)*sin(Δλ/2)**2
    return R * 2 * asin(sqrt(a))

17 AI-Native Architecture — How It Works

The closed loop

The OMS has a continuous feedback loop between order execution and sourcing intelligence. Every sourcing decision is an experiment. Every delivery is a data point. The system accumulates evidence, identifies patterns, and proposes improvements — all without touching any existing configuration until a human approves.

Design principles

Layer 1 — Intelligence data foundation

The entire learning system is built on one collection: sourcing_outcomes in the oms_ai_learning MongoDB database. Every time the sourcing engine assigns an order to a node, it writes a document capturing a snapshot of the decision. This database is separate from oms_events so that nightly aggregation queries do not contend with transactional order audit reads.

sourcing_outcomes document lifecycle

Outcome score formula

outcome_score = (
    0.4 * delivery_score    # 1.0 if ≤24h | 0.5 if ≤48h | 0.2 if ≤72h | 0.0 if >72h
  + 0.3 * cost_score        # 1.0 if variance ≤5% | 0.7 if ≤15% | 0.3 if ≤25% | 0.0 if >25%
  + 0.2 * (1 - backordered) # 0.2 if no backorder, 0.0 if backordered
  + 0.1 * (1 - returned)    # 0.1 if not returned, 0.0 if returned
)

A perfect sourcing decision (delivered in under 24 hours, cost within 5% of estimate, no backorder, no return) scores 1.0. A bad one (72+ hour delivery, 25%+ cost overrun, backorder, and return) scores 0.0.

Layer 2 — AI sourcing engine

When a sourcing rule uses AI_ADAPTIVE or AI_HYBRID strategy, the AISourcingAdvisor is invoked before the final node is chosen.

What KubeAI receives

The advisor builds a structured prompt containing:

• Order context — channel, fulfillment type, customer region, order amount, SKU count.
• Historical patterns — the top matching cluster(s) for this order type, with node rankings and sample counts. Up to 3 clusters are included (exact match first, then broader channel+fulfillment_type matches).
• Node performance — rolling 7-day metrics for each candidate node: orders fulfilled, avg outcome score, avg delivery hours, backorder rate.
• Candidates — each eligible node with its distance, estimated cost, and available inventory.

Score blending

# AI_ADAPTIVE: KubeAI score weighted 60%, rule-based score 40%
final_score = 0.6 * ai_score + 0.4 * rule_score

# AI_HYBRID: same formula, different label in sourcing_outcomes
# (useful for distinguishing "AI primary" vs "transitional AI" in analytics)

Fallback triggers

Fallback decisions are visible in the order's sourcing decision trail under rule_details.ai_sourcing.fallback_used and fallback_reason.

Layer 3 — Pattern discovery

The PatternDiscoveryService runs nightly (Celery beat, 02:00 UTC) via the discover_patterns task in the learning queue.

Step 1 — Aggregate patterns

MongoDB $group aggregation on sourcing_outcomes where outcome_score exists, grouped by (cluster_key, node_id). For each cluster, nodes are ranked by avg_outcome_score and upserted into sourcing_patterns.

Step 2 — Compare strategies

A second aggregation groups by (cluster_key, strategy_used) to get avg outcome scores per strategy per cluster. Qualifying clusters must meet all three thresholds:

• Total samples (AI + baseline) ≥ 50
• AI_ADAPTIVE samples ≥ 10
• AI improvement over DISTANCE_OPTIMAL ≥ 10%

Step 3 — Generate proposals

For each qualifying cluster, one pending AIProposal is created — but only if no active (PENDING or APPROVED) proposal already exists for that cluster key. This prevents duplicate proposals. The proposal includes auto-generated sourcing rule conditions derived from the cluster dimensions (channel, region, amount range, fulfillment type).

Layer 4 — A/B experiments

Experiments allow side-by-side comparison of any two sourcing strategies on real orders without committing to a permanent rule change. See §15 AI Architect for the full experiment usage guide.

Experiment evaluation (daily 03:00 UTC)

The evaluate_ai_experiments task checks all running experiments. Winner declaration requires:

• Each arm has ≥ 50 labeled outcomes (orders that have reached DELIVERED)
• Absolute outcome score difference between arms ≥ 0.05

If neither arm has a clear edge, the experiment continues until both thresholds are met or an admin manually stops it.

Layer 5 — Node performance metrics

The update_node_performance task runs every 4 hours. It aggregates sourcing_outcomes over the last 7 and 30 days per node, computing: orders fulfilled, avg outcome score, avg delivery hours, avg actual cost, backorder rate %, return rate %. These metrics feed into the AI sourcing prompt as the "NODE PERFORMANCE" context block.

MongoDB collections reference

All three AI learning collections live in oms_ai_learning (config key: MONGODB_AI_DB), separate from the oms_events transactional database.

PostgreSQL tables reference

Celery learning tasks

18 Fulfillment Pipeline

Order Status State Machine

Fulfillment Types & Required Capabilities

19 Celery Workers

Celery Beat Schedule

Start Workers

celery -A app.workers.celery_app worker \
  --loglevel=info \
  -Q sourcing,fulfillment,carrier,notifications,webhooks,connectors,learning \
  --concurrency=4

21 Webhook System

HMAC-SHA256 Signing

Every outbound webhook request is cryptographically signed. Receivers can verify payload integrity:

POST https://your-endpoint.com/hooks HTTP/1.1
Content-Type: application/json
X-OMS-Signature: sha256=abc123...
X-OMS-Timestamp: 1708000000
X-OMS-Event: order.shipped

Verification (Python)

import hmac, hashlib

def verify_webhook(body: bytes, signature: str, secret: str) -> bool:
    expected = "sha256=" + hmac.new(
        secret.encode(), body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

Supported Event Types

Retry Strategy (Exponential Backoff)

22 Connector System

The Connector System enables bidirectional integration between the OMS and external platforms — e-commerce engines, carriers, WMS, and TMS systems. Each connector has a configurable direction (inbound, outbound, or bidirectional) and handles its own authentication protocol.

Architecture

Shopify                              OMS (single URL per connector)
────────                             ─────────────────────────────
orders/create  ──┐
orders/paid    ──┼──► POST /connectors/{id}/webhook
products/create──┤        │ HMAC-SHA256 validated
products/update──┘        │ routes by X-Shopify-Topic header
                          │
             ┌────────────┴────────────────────────┐
             │                                      │
         Order topics                         Product topics
             │                                      │
             ▼                                      ▼
 normalize_order() → OMS Order         normalize_product() → InventoryItems
 (channel=MARKETPLACE,                 (upsert per active node,
  connector_id set, dedup check)        seed stock from Shopify qty)
             │
             ▼  (when order.status → SHIPPED)
 Celery: sync_fulfillment_to_connector
             │
             ▼
 push_fulfillment() → POST /orders/{shopify_id}/fulfillments
             │
             ▼
 Shopify notifies buyer of tracking info

Supported Platforms

Shopify Setup

1. In the OMS Admin Console → Connectors, click Add Connector and select Shopify.

2. Enter your Shopify store URL, Admin API access token, and generate a webhook secret.

3. Copy the webhook URL displayed in the connector card:

https://your-oms.example.com/connectors/{connector_id}/webhook

4. In Shopify Admin → Settings → Notifications → Webhooks, register the same URL for each topic you want to receive. All topics share one URL and one signing secret:

Order topics → import as OMS orders. Product topics → upsert InventoryItem records across all active fulfillment nodes.

5. Enable the connector (toggle to Active). Orders and product catalog changes from Shopify will automatically sync into the OMS.

Amazon SP-API Setup

Amazon uses polling, not webhooks. The OMS polls the SP-API every 15 minutes automatically via a Celery Beat scheduled task — no webhook registration is required.

1. In Seller Central, navigate to Apps & Services → Develop Apps. Create a new app to obtain your client_id and client_secret.

2. Generate a refresh_token by completing the OAuth authorization flow for your selling account.

3. In the OMS Admin Console → Connectors, click Add Connector and select Amazon SP-API. Provide the following fields:

4. Click Test Connection. The OMS calls GET /sellers/v1/marketplaceParticipations to verify credentials and retrieve marketplace details.

5. Enable the connector. The Celery Beat schedule (poll_amazon_orders) will automatically pull new orders every 15 minutes and import them as OMS orders with channel=MARKETPLACE.

Multi-Topic Support (Same URL)

A single connector webhook URL handles every event topic Shopify sends to it. The OMS routes by the X-Shopify-Topic header after HMAC validation:

Inbound Webhook Security

Connector webhook endpoints (POST /connectors/{id}/webhook) are exempt from JWT authentication and instead validated using platform-native HMAC signatures.

# Shopify HMAC validation (inside ShopifyConnector)
import hmac, hashlib, base64

def validate_webhook(self, headers: dict, raw_body: bytes) -> bool:
    hmac_header = headers.get("x-shopify-hmac-sha256", "")
    secret = self.config.get("webhook_secret", "")
    digest = base64.b64encode(
        hmac.new(secret.encode(), raw_body, hashlib.sha256).digest()
    ).decode()
    return hmac.compare_digest(digest, hmac_header)

Outbound Fulfillment Sync

When an OMS order (sourced from a connector) transitions to SHIPPED status, a Celery task is automatically enqueued on the connectors queue to push the tracking update back to the originating platform.

# Triggered automatically from PATCH /orders/{id}/status
if payload.status == OrderStatus.SHIPPED and order.connector_id:
    background_tasks.add_task(_trigger_connector_sync, str(order.id))

Outbound Sync — OMS → Platform

When the OMS changes an order's state or adjusts inventory, it pushes the update back to the originating platform. Each push is handled by a dedicated Celery task on the connectors queue.

Shopify

Amazon SP-API

Extending with New Connectors

All connectors inherit from BaseConnector in app/services/connectors/base.py. Four methods are abstract (required); two are optional overrides for product/catalog sync:

class BaseConnector(ABC):
    # ── Required (abstract) ──────────────────────────────────────
    def validate_webhook(self, headers, raw_body) -> bool: ...
    def get_event_type(self, headers) -> str: ...
    def normalize_order(self, payload) -> dict: ...       # → OMS OrderCreate dict
    async def push_fulfillment(self, order, shipment) -> dict: ...
    async def test_connection(self) -> dict: ...

    # ── Optional (default: empty set / empty list) ────────────────
    def get_inbound_topics(self) -> set[str]: ...         # order-creation topics
    def get_product_topics(self) -> set[str]: ...         # product-sync topics
    def normalize_product(self, payload) -> list[dict]: ...  # → variant dicts

    # ── Outbound sync (default: no-op) ───────────────────────────
    async def push_inventory_update(self, sku, quantity_available, mapping) -> dict: ...
    async def push_order_cancel(self, order) -> dict: ...

Register the new class in app/services/connectors/registry.py to make it available.

Event Audit Log

Every inbound and outbound connector activity is recorded in the connector_events table:

23 Monitoring & Traceability

The OMS has a built-in error monitoring console accessible at Admin → Monitoring. Every unhandled exception across all services — FastAPI routes, Celery workers, background tasks — is automatically captured, fingerprinted, and stored in MongoDB for analysis.

Error Capture Pipeline

Exception raised (anywhere)
        │
        ▼
capture_error() / capture_error_sync()   ← app/services/monitoring.py
        │
        ├─► error_events (MongoDB)   — one document per occurrence, 30-day TTL
        │     event_id, fingerprint, timestamp, level, source_service,
        │     error_type, error_message, stack_frames,
        │     request_context, task_context, order_context
        │
        └─► error_issues (MongoDB)   — one document per unique fingerprint
              fingerprint, status (open/resolved/muted),
              occurrence_count, first_seen_at, last_seen_at

Source Services

Monitoring API /monitoring

All endpoints require superadmin authentication.

Fingerprinting

Each error is assigned a stable 16-character fingerprint so that repeated occurrences of the same error (e.g. a recurring database connection timeout) are grouped into a single issue rather than creating noise:

fingerprint = SHA256(f"{source_service}:{error_type}:{top_stack_frame}")[:16]

The top stack frame is the deepest frame inside application code (excluding library internals), extracted from the Python traceback.

Issue Lifecycle

Verifying the Pipeline

curl -X POST http://localhost:8001/monitoring/test-error \
  -H "Authorization: Bearer {superadmin_token}" \
  -H "Content-Type: application/json" \
  -d '{"message": "Smoke test", "source": "api", "level": "ERROR"}'

After calling this endpoint, navigate to Monitoring → Issues or Monitoring → Events in the UI to confirm the entry appears.

24 Configuration

All configuration is managed via environment variables. For local development, create a .env file in the project root.

25 Running the System

Start All Services

cd D:\OMS
docker compose up -d --build

Services started after this command:

Seed All Databases

docker compose exec api python scripts/seed.py

Seeds the following data:

• PostgreSQL: 8 fulfillment nodes, 64 inventory items (8 SKUs × 8 nodes), 5 sourcing rules, 1 webhook endpoint
• MongoDB: 8 product catalog documents, 3 sample order events
• Redis: version, stats, and cache warmup keys
• Elasticsearch: 8 product documents, 3 sample order documents

Service URLs

Run Tests

PYTHONPATH=D:\OMS python -m pytest tests/test_imports.py -v

26 End-to-End Order Flow

Step 1 — Create an Order

curl -X POST http://localhost:8001/orders/ \
  -H "Content-Type: application/json" \
  -d '{
    "channel": "WEB",
    "fulfillment_type": "SHIP_TO_HOME",
    "customer_email": "customer@example.com",
    "customer_name": "John Doe",
    "line_items": [
      {"sku": "SKU-WIDGET-A", "product_name": "Premium Widget A",
       "quantity": 2, "unit_price": 29.99},
      {"sku": "SKU-GADGET-X", "product_name": "Gadget X Pro",
       "quantity": 1, "unit_price": 99.99}
    ],
    "shipping_address": {
      "address1": "456 Park Ave", "city": "New York", "state": "NY",
      "postal_code": "10022", "latitude": 40.7614, "longitude": -73.9776
    }
  }'

Response (HTTP 201): Returns the new order with "status": "PENDING" and a generated order_number like ORD-20240215-XK9M2A.

Step 2 — Sourcing Fires Automatically (~2 seconds)

1. Celery sourcing worker picks up the source_order task
2. Loads all active sourcing rules sorted by priority
3. Evaluates the catch-all "Default — Distance Optimal" rule
4. Loads all ACTIVE nodes with inventory for each requested SKU
5. Computes haversine distance from customer coordinates to each node
6. Scores nodes — STR-NYC-01 at ~1.8km wins for a NYC customer
7. Creates FulfillmentAllocation rows (one per SKU)
8. Reserves inventory: quantity_available -= quantity_allocated
9. Order transitions to SOURCED

Step 3 — Fulfillment Pipeline (~5–10 seconds)

• Picking (t+2s): Allocations → PICKING, order → PICKING
• Packing (t+7s): Allocations → PACKED, order → PACKING → READY_TO_SHIP

Step 4 — Carrier Booking

• Selects a random carrier (UPS, FedEx, USPS, DHL)
• Generates a mock tracking number
• Creates a Shipment record with estimated delivery date
• Order → SHIPPED, allocations → SHIPPED
• Fires order.shipped notification + webhook

Step 5 — Delivery Simulation (~10 seconds later)

Adds tracking events: IN_TRANSIT → OUT_FOR_DELIVERY → DELIVERED. Order → DELIVERED. Fires order.delivered webhook.

Step 6 — Verify via Search

curl -X POST http://localhost:8001/search/orders \
  -H "Content-Type: application/json" \
  -d '{"query": "John Doe", "status": "DELIVERED"}'

Step 7 — Check Audit Trail

curl http://localhost:8001/orders/{order_id}/events

Returns chronological MongoDB events: order.created → order.sourced → order.shipped → order.delivered

Step 8 — Analytics

curl "http://localhost:8001/analytics/dashboard?from_date=2024-01-01"

27 Seed Data Reference

Fulfillment Nodes (8 total)

Sourcing Rules (5 active)

Product SKUs (8 SKUs × 8 nodes = 64 inventory records)

28 Admin Console

The Admin Console (/admin) is the central hub for user management, permission configuration, and access control across the entire platform. It is accessible only to users with SUPERADMIN or PLATFORM_OWNER platform roles.

RBAC Model — Three-Layer Hierarchy

Access in the OMS is controlled by three layers that stack on top of each other:

Tab 1 — Users

Shows all users registered on the platform. Each row displays the username, email, platform role badge, group membership, account status (active / suspended), and the last login time.

Platform Role Management

Click any user row to open the User Access Drawer — a right-side panel showing the user's complete access picture across all organizations and environments. From the drawer you can:

• Change platform role — upgrade/downgrade between PLATFORM_OWNER, SUPERADMIN, USER.
• Grant org role — assign the user an organization-level role (ORG_OWNER / ORG_ADMIN / ORG_MEMBER) for any existing organization. This gives the user visibility into all environments within that org.
• Revoke org role — removes the user's access to all environments within that organization (unless they also have individual env roles).
• Grant env role — assign a specific role (OWNER / ADMIN / MEMBER / VIEWER) for a specific environment, regardless of org membership.
• Revoke env role — removes the user's individual access to that environment.

Creating a User with Org/Environment Access

To add a new user and grant them access to a specific organization and environment:

1. Create the user account

   In the Users tab, click Create User. Fill in name, email, password, platform role (USER for standard access), and optionally assign a group.

2. Open the User Access Drawer

   Click the user row in the Users table. The drawer opens on the right side.

3. Grant an Org Role (optional)

   In the Organization Access section, click Grant Org Role. Select the organization and role (ORG_MEMBER for read-only, ORG_ADMIN to let them manage environments). This lets the user see the org in their environment switcher.

4. Grant an Env Role

   In the Environment Access section, click Grant Env Role. Select the specific environment and role (ADMIN to manage orders/inventory, MEMBER for day-to-day operations, VIEWER for read-only). The user can now switch to this environment using the environment switcher.

Tab 3 — Access Control

A higher-level view showing who has access to each organization and environment, organized by the tenant hierarchy rather than by user. Useful for auditing who can access what without going user-by-user.

The tab has two sub-tabs:

• Org View — shows each organization with a table of all users who have an org-level role. Use Add Member to directly grant an org role here. Use the trash icon to revoke it.
• Env View — shows each environment (grouped by org) with its member list. Use Add Member to grant an env role. The environment type badge (DEV/QA/STAGING/PROD) is color-coded for quick scanning.

Granting access directly from Access Control

1. Switch to "Org View" or "Env View"

   Use the sub-tabs at the top of the Access Control tab.

2. Find the org or environment

   Each card shows the org/env name and the current member list.

3. Click "Add Member"

   A modal appears. Start typing a name or email to search for existing users. Select the user and pick a role using the radio buttons.

4. Confirm

   Click Grant Access. The member list updates immediately. The user will see the environment in their switcher on next page load.

Tab 2 — Groups & Permissions

Permission groups let you define reusable sets of feature permissions that can be assigned to users. A permission is a feature:action string, e.g. orders:write, inventory:read, connectors:manage.

Tab 4 — Access Matrix

A grid view showing all users (rows) × all environments (columns). Each cell displays the user's effective role for that environment — considering both their org role and their direct env role. Empty cells mean the user has no access to that environment.

This view is ideal for a quick compliance audit: you can see at a glance whether any user has unexpectedly broad access (e.g. OWNER on the Production environment) and immediately open their drawer to adjust.

Multi-Tenant Platform Architecture

The OMS is a multi-tenant, multi-environment platform. A single infrastructure deployment serves multiple client organizations, each with fully isolated data environments — their own PostgreSQL databases, MongoDB databases, and Elasticsearch index namespaces. No data ever crosses organizational boundaries at the storage layer.

Key design principles

29a Role Hierarchy

Platform roles (global — stored on users.platform_role)

Organization roles (per org — user_organization_roles)

Environment roles (per environment — user_environment_roles)

29b Creating an Organization — Step by Step

An organization is a logical tenant container. No database is provisioned at this point — it is a pure control-plane record. Creating one is instantaneous.

Via the UI (Platform Console)

1. Navigate to Platform Console

   In the sidebar, click Platform. Visible only to PLATFORM_OWNER and SUPERADMIN users. Shows all organizations and a summary of their environments.

2. Click “New Organization”

   Fill in Name (e.g. Acme Corp), Slug (e.g. acme — globally unique, pattern ^[a-z0-9][a-z0-9-]*[a-z0-9]$), and optional description. The slug becomes the prefix for all database names in this org.

3. Submit

   POST /organizations → 201 Created. Org appears in the list with zero environments. No database has been created yet.

4. Add the first environment

   Click the org card, then New Environment. See the provisioning section below.

Via the API

curl -X POST http://localhost:8001/organizations \
  -H "Authorization: Bearer {platform_owner_token}" \
  -H "Content-Type: application/json" \
  -d '{"name": "Acme Corp", "slug": "acme", "description": "Our first tenant"}'

Slug naming convention

Organizations API

29c Provisioning an Environment — Step by Step

An environment is where business data actually lives — orders, inventory, sourcing rules, connectors. Creating one triggers a full infrastructure provisioning sequence that runs asynchronously. The API returns 202 Accepted with status: "PROVISIONING"; the UI polls every 5 seconds until status becomes "ACTIVE".

Via the UI (Environments page)

1. Open the Environments page

   Click Environments in the sidebar. Lists all environments the current user can access, grouped by organization.

2. Click “New Environment”

   Fill in: Organization, Name (e.g. Development), Slug (e.g. dev, unique within org), Environment Type (DEV/QA/STAGING/PROD), and optionally Set as default.

3. Submit — provisioning begins

   Environment created with status: PROVISIONING. Background task starts. UI shows a spinner polling every 5 seconds.

4. Wait for ACTIVE status

   Typical provisioning time: 5–15 seconds. The environment card turns green when status = ACTIVE and shows the provisioned_at timestamp.

5. Generate Docker Compose file (for isolated deployment)

   Click Generate Docker Compose on the environment card. Downloads docker-compose.acme_dev.yml pre-configured with the correct database names, ports, and network settings.

6. Set Deployment URL

   If the environment runs as a separate pod, set the Deployment URL field (e.g. http://localhost:3463). The environment switcher uses this for cross-origin pod redirects with seamless auth handoff.

Provisioning sequence (what happens in the background)

Step 1 -- Create PostgreSQL database
  Raw asyncpg connection to postgres system DB
  CREATE DATABASE oms_acme_dev;   (idempotent -- checks existence first)

Step 2 -- Build SQLAlchemy engine
  create_async_engine("postgresql+asyncpg://.../oms_acme_dev",
    pool_size=5, max_overflow=10, pool_pre_ping=True, pool_recycle=3600)
  Register: EnvironmentEngineRegistry._engines[env_id] = engine

Step 3 -- Create all data-plane tables  (checkfirst=True -- skips if exists)
  orders, order_items, fulfillment_allocations, shipments
  inventory_items, inventory_adjustments, inventory_reservations
  fulfillment_nodes, sourcing_rules
  webhook_endpoints, webhook_events
  connectors, connector_events, connector_inventory_mappings
  ai_proposals, ai_experiments
  sourcing_outcome_labels, custom_attribute_definitions, ui_widgets

  NOT created here (stay in oms_db only):
  users, user_groups, organizations, environments
  user_environment_roles, user_organization_roles

Step 4 -- Create MongoDB indexes
  oms_events_acme_dev:
    order_events -> compound (order_id, timestamp), index on event_type
  oms_ai_acme_dev:
    sourcing_outcomes -> order_id index
    sourcing_patterns -> unique cluster_key index

Step 5 -- Create Elasticsearch index template
  PUT /_index_template/acme_dev_template
  {"index_patterns": ["acme_dev_*"], "mappings": {...}}

Step 6 -- Mark environment ACTIVE
  UPDATE environments SET status='ACTIVE', provisioned_at=now()
  WHERE id = <env_id>   (committed on control DB oms_db)

Connection pool per environment

pool_size=5        # 5 persistent connections
max_overflow=10    # burst up to 15 total
pool_pre_ping=True # discard stale connections
pool_recycle=3600  # recycle every hour

20 active environments = at most 300 PostgreSQL connections total — well within default max_connections=100–500.

Environments API

29d Environment Types & Lifecycle

Environment status lifecycle

29e Managing Environment Members

Grant org membership (gives org-wide visibility)

curl -X POST http://localhost:8001/organizations/{org_id}/members \
  -H "Authorization: Bearer {admin_token}" \
  -H "Content-Type: application/json" \
  -d '{"user_id": "abc123...", "role": "ORG_MEMBER"}'

Grant environment access

curl -X POST http://localhost:8001/admin/users/{user_id}/env-roles \
  -H "Authorization: Bearer {admin_token}" \
  -H "Content-Type: application/json" \
  -d '{"env_id": "def456...", "role": "ADMIN"}'

View a user's full access

GET /admin/users/{user_id}/access
{
  "user_id": "...",
  "platform_role": "USER",
  "org_roles": [
    {"org_id": "...", "org_name": "Acme Corp", "role": "ORG_MEMBER"}
  ],
  "env_roles": [
    {"env_id": "...", "env_name": "Development",
     "org_name": "Acme Corp", "env_type": "DEV", "role": "ADMIN"}
  ]
}

Admin Console access management endpoints

29f Environment Switcher & Cross-Pod Auth

Switching environments

The Environment Switcher lives in the top header bar, grouped by organization. On switch, the frontend:

1. Stores the selected env_id in localStorage('oms_environment_id')
2. Calls queryClient.clear() to flush all cached data from the previous environment
3. Re-fetches all active queries with the new environment header
4. If the environment has a base_url on a different origin, performs a cross-origin redirect with token handoff (no re-login)

How X-OMS-Environment works

axiosInstance.interceptors.request.use(config => {
  const envId = localStorage.getItem('oms_environment_id')
  if (envId) config.headers['X-OMS-Environment'] = envId
  const token = localStorage.getItem('oms_auth_token')
  if (token) config.headers['Authorization'] = `Bearer ${token}`
  return config
})

Cross-pod auth handoff (no re-login on switch)

;(function bootstrapFromUrl() {
  const params = new URLSearchParams(window.location.search)
  const token   = params.get('_oms_token')
  const userRaw = params.get('_oms_user')
  if (token && userRaw) {
    localStorage.setItem('oms_auth_token', token)
    localStorage.setItem('oms_auth_user', userRaw)
    params.delete('_oms_token'); params.delete('_oms_user')
    const clean = params.toString()
    window.history.replaceState({}, '', window.location.pathname +
      (clean ? '?' + clean : '') + window.location.hash)
  }
})()

29g Deploying a Tenant Environment (Docker Compose)

Each environment can run as an isolated Docker Compose stack — its own API and frontend containers — attached to the shared infrastructure via the oms_default external network.

Example: docker-compose.acme_dev.yml

version: "3.9"

services:
  api_acme_dev:
    build: { context: ., dockerfile: Dockerfile }
    container_name: oms_api_acme_dev
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000
    environment:
      - TENANT_SLUG=acme
      - ENVIRONMENT=dev
      - DATABASE_URL=postgresql+asyncpg://oms_user:oms_pass@postgres:5432/oms_acme_dev
      - SYNC_DATABASE_URL=postgresql+psycopg2://oms_user:oms_pass@postgres:5432/oms_acme_dev
      - CONTROL_DATABASE_URL=postgresql+asyncpg://oms_user:oms_pass@postgres:5432/oms_db
      - MONGODB_DB=oms_events_acme_dev
      - MONGODB_AI_DB=oms_ai_acme_dev
      - FRONTEND_URL=http://localhost:3463
    ports: ["8463:8000"]
    networks: [oms_default]

  frontend_acme_dev:
    build: { context: ./frontend, dockerfile: Dockerfile }
    container_name: oms_frontend_acme_dev
    environment:
      - API_HOST=oms_api_acme_dev
    ports: ["3463:80"]
    networks: [oms_default]
    depends_on: [api_acme_dev]

networks:
  oms_default:
    external: true
    name: oms_default

Start the tenant stack

docker-compose -f docker-compose.acme_dev.yml up -d --build

# Then set the Deployment URL:
# Platform Console > Environments > Acme Dev > Deployment URL = http://localhost:3463

Delete an environment (with data)

1. Stop the tenant stack

   docker-compose -f docker-compose.acme_dev.yml down

2. Delete from the UI

   Environments page → trash icon on the environment card → confirm. Calls DELETE /environments/{id}, which drops the PostgreSQL database and removes all control-plane records. This is irreversible.

3. Remove the Compose file

   rm docker-compose.acme_dev.yml

29h How Requests Are Routed to the Right Database

Every API request carries an X-OMS-Environment HTTP header. The EnvironmentMiddleware resolves it to the correct database before any route handler runs.

Browser
  X-OMS-Environment: <env_uuid>
  |
EnvironmentMiddleware (app/middleware/environment.py)
  |
  +-- Exempt paths (bypass resolution):
  |   /health, /docs, /redoc, /openapi.json
  |   /auth/*, /testing, /connectors/{id}/webhook
  |
  +-- 1. Redis: GET env:<env_uuid>  TTL=60s
  |      HIT  -> deserialize cached env object (lock-free hot path)
  |      MISS -> SELECT * FROM environments WHERE id=?  (control DB)
  |             cache -> SET env:<env_uuid> EX 60
  |
  +-- 2. env.status != ACTIVE -> return 503 "Environment unavailable"
  |
  +-- 3. request.state.environment = env -> call_next(request)

Route handler: db = Depends(get_db)
  |
  +-- env.db_name == "oms_db"  (default production)
  |    -> yield from main session factory
  |
  +-- env-specific database
       -> registry.get_or_create_engine(env)  (lock-free if already cached)
       -> yield session from env-specific session factory

All SQL runs against oms_acme_dev. Zero code changes in business logic.

Which routes use which database

29i Data Isolation & Backward Compatibility

Database-per-environment isolation

Row-level security (shared table + tenant_id column)
  > Weakest. A missing WHERE clause leaks all tenant data.
  > OMS does NOT use this.

Schema-per-tenant (shared DB, different schemas)
  > Medium. Connection pool shared; search_path misconfiguration leaks data.
  > OMS does NOT use this.

Database-per-tenant  <--  OMS uses this for each environment
  > Strongest. Separate connection pool, separate WAL, separate backups.
  > No SQL query can cross the database boundary.

Backward compatibility — zero breaking changes

The first startup seeds a "Default Organization" with a "Production" environment pointing at the existing oms_db database. All existing data (orders, inventory, nodes, sourcing rules) is untouched. Requests without an X-OMS-Environment header fall back to the environment where is_default=true, which points at oms_db. Existing JWT tokens and API clients all continue to work.

Celery — environment-aware fan-out

@celery_app.task
def source_pending_orders_fanout():
    # Beat fires this every 2 min. One task per active environment.
    env_ids = list_active_environment_ids()   # sync SQL on oms_db
    for env_id in env_ids:
        source_pending_orders.delay(env_id)

@celery_app.task
def source_pending_orders(environment_id: str = ""):
    # Empty env_id = backward-compat fallback to settings.DATABASE_URL
    db_url = get_env_db_url(environment_id)
    # ... unchanged sourcing logic ...

MongoDB & Elasticsearch per environment