Marketplace Connector Architecture: Patterns from Building 35+ Integrations

Why Architecture Matters More Than Code

When I built my first marketplace connector in 2019, I focused on getting the code right: proper API calls, correct data mapping, thorough testing. The connector worked perfectly in development. Then it hit production, where Amazon sent duplicate webhooks, network timeouts corrupted partial syncs, and a flash sale generated 800 orders in 20 minutes that overwhelmed a synchronous processing pipeline.

That experience taught me that connector architecture determines success or failure far more than individual code quality. Thirty-five connectors later, I have a set of patterns that I apply to every integration project. This post covers the architecture I wish someone had shown me before I learned these lessons the hard way.

The Core Architecture

Every marketplace connector I build follows the same high-level architecture, regardless of whether it is Amazon, Shopify, eBay, Walmart, Daraz, or a regional platform.

Marketplace API
       |
       v
  [Ingest Layer]     ← Webhooks + Polling
       |
       v
  [Message Queue]    ← RabbitMQ / Redis Streams
       |
       v
  [Processing Layer] ← Workers with retry logic
       |
       v
  [ERP Adapter]      ← Odoo XML-RPC / REST API
       |
       v
  [Reconciliation]   ← Nightly validation

Let me walk through each layer and the decisions behind it.

Layer 1: The Ingest Layer

The ingest layer has one job: accept incoming data and put it on the queue. It does not process, validate, or transform. This separation is critical.

Webhook Handler

from fastapi import FastAPI, Request, HTTPException
from datetime import datetime
import hashlib
import hmac

app = FastAPI()

@app.post("/webhooks/{marketplace}/{event_type}")
async def ingest_webhook(
    marketplace: str,
    event_type: str,
    request: Request
):
    body = await request.body()

    # Verify signature (marketplace-specific)
    if not verify_signature(marketplace, request.headers, body):
        raise HTTPException(status_code=401)

    # Generate idempotency key
    idempotency_key = generate_idempotency_key(marketplace, body)

    # Check for duplicate
    if await is_duplicate(idempotency_key):
        return {"status": "duplicate", "key": idempotency_key}

    # Enqueue with metadata
    await queue.publish(
        channel=f"{marketplace}.{event_type}",
        message={
            "idempotency_key": idempotency_key,
            "marketplace": marketplace,
            "event_type": event_type,
            "payload": body.decode(),
            "received_at": datetime.utcnow().isoformat(),
            "headers": dict(request.headers),
        }
    )

    return {"status": "accepted", "key": idempotency_key}

Polling Service

For marketplaces that do not support webhooks (or where webhooks are unreliable), a polling service runs on a schedule:

class MarketplacePoller:
    def __init__(self, marketplace_client, queue, state_store):
        self.client = marketplace_client
        self.queue = queue
        self.state = state_store

    async def poll_orders(self):
        # Get last sync timestamp
        last_sync = await self.state.get("last_order_sync")

        # Fetch new/updated orders since last sync
        orders = await self.client.get_orders(
            updated_after=last_sync,
            status=["paid", "partially_fulfilled"]
        )

        for order in orders:
            idempotency_key = f"order-{order['id']}-{order['updated_at']}"

            if not await is_duplicate(idempotency_key):
                await self.queue.publish(
                    channel="orders.sync",
                    message={
                        "idempotency_key": idempotency_key,
                        "order": order,
                        "source": "poll"
                    }
                )

        # Update sync timestamp
        await self.state.set(
            "last_order_sync",
            datetime.utcnow().isoformat()
        )

The idempotency key includes the updated_at timestamp. This means the same order polled twice at the same state is deduplicated, but a genuinely updated order gets processed again.

Layer 2: The Message Queue

I use RabbitMQ for most production connectors. Redis Streams work for simpler setups. The queue provides three essential capabilities:

Decoupling

The ingest layer and processing layer run independently. If the processing layer goes down for maintenance, webhooks still get accepted and queued. When processing resumes, it catches up automatically.

Backpressure

During traffic spikes (flash sales, Prime Day, 11.11), the queue absorbs the burst. Workers process at their own pace without overwhelming the ERP. Without a queue, a sudden spike can cascade into ERP timeouts, failed API calls, and data loss.

Retry Isolation

Failed messages can be retried independently. A single malformed order does not block the processing of a thousand valid ones.

Queue Design Decisions

# Queue topology I use for each marketplace
QUEUES = {
    # Separate queues by data type for independent scaling
    "orders.create": {"workers": 4, "prefetch": 10},
    "orders.update": {"workers": 2, "prefetch": 10},
    "orders.cancel": {"workers": 1, "prefetch": 5},
    "inventory.update": {"workers": 2, "prefetch": 50},
    "products.sync": {"workers": 1, "prefetch": 20},
    "fulfillment.push": {"workers": 2, "prefetch": 10},

    # Dead letter queue for failed messages
    "dlq": {"workers": 0, "prefetch": 0},  # Manual processing only
}

Separate queues per data type let you scale independently. Order processing might need 4 workers during peak hours while product sync needs only 1.

Layer 3: The Processing Layer

This is where data transformation, business logic, and ERP interaction happen. Each worker follows the same pattern:

class OrderProcessor:
    def __init__(self, erp_adapter, marketplace_client):
        self.erp = erp_adapter
        self.marketplace = marketplace_client

    async def process(self, message: dict) -> ProcessingResult:
        order_data = message["payload"]

        try:
            # Step 1: Check idempotency (belt and suspenders)
            existing = await self.erp.find_order_by_external_id(
                order_data["id"]
            )
            if existing:
                return ProcessingResult.DUPLICATE

            # Step 2: Resolve references
            customer = await self.resolve_customer(order_data)
            line_items = await self.resolve_products(order_data["items"])

            # Step 3: Create ERP record
            erp_order = await self.erp.create_sale_order(
                customer=customer,
                lines=line_items,
                external_id=order_data["id"],
                marketplace=message["marketplace"]
            )

            # Step 4: Log the mapping
            await self.log_mapping(
                external_id=order_data["id"],
                internal_id=erp_order.id,
                marketplace=message["marketplace"]
            )

            return ProcessingResult.SUCCESS

        except RetryableError as e:
            # Network timeout, API rate limit, temporary ERP unavailability
            raise  # Let the queue retry mechanism handle it

        except PermanentError as e:
            # Invalid data, missing required fields, business rule violation
            await self.send_to_dlq(message, error=str(e))
            return ProcessingResult.FAILED

The Critical Distinction: Retryable vs Permanent Errors

This is the single most important error handling decision in connector architecture. Get it wrong and you either lose data (treating retryable errors as permanent) or create infinite retry loops (treating permanent errors as retryable).

Retryable errors (retry with exponential backoff):

• Network timeouts
• HTTP 429 (rate limited)
• HTTP 500/502/503 from the marketplace or ERP
• Database connection failures
• Lock contention

Permanent errors (send to dead letter queue):

• HTTP 400 (bad request) — the data is malformed
• Missing required fields that cannot be resolved
• Business rule violations (duplicate order, invalid product reference)
• Authentication failures (credentials expired)

class RetryPolicy:
    MAX_RETRIES = 5
    BASE_DELAY = 1  # seconds

    @staticmethod
    def calculate_delay(attempt: int) -> float:
        # Exponential backoff with jitter
        delay = RetryPolicy.BASE_DELAY * (2 ** attempt)
        jitter = random.uniform(0, delay * 0.1)
        return min(delay + jitter, 300)  # Cap at 5 minutes

Layer 4: The ERP Adapter

The adapter translates between the connector's internal data model and the ERP's API. This layer is the only part of the system that knows about Odoo (or whatever ERP you are targeting).

class OdooAdapter:
    """Abstracts all Odoo-specific logic"""

    async def create_sale_order(self, customer, lines, external_id, marketplace):
        return await self.odoo.create("sale.order", {
            "partner_id": customer.odoo_id,
            "order_line": [
                (0, 0, {
                    "product_id": line.odoo_product_id,
                    "product_uom_qty": line.quantity,
                    "price_unit": line.unit_price,
                })
                for line in lines
            ],
            "x_external_id": external_id,
            "x_marketplace": marketplace,
            "x_sync_status": "synced",
        })

    async def find_order_by_external_id(self, external_id):
        results = await self.odoo.search_read(
            "sale.order",
            [("x_external_id", "=", external_id)],
            ["id", "name", "state"]
        )
        return results[0] if results else None

Why abstract the ERP? Because I have had clients switch from Odoo to NetSuite mid-project. With a clean adapter layer, we swapped out the adapter without touching the processing logic. This has happened three times across 35 projects. The abstraction pays for itself.

Layer 5: Reconciliation

No integration is perfect. Network issues, timing windows, and edge cases mean that data will occasionally get out of sync. The reconciliation layer catches these discrepancies before they become business problems.

class DailyReconciler:
    async def reconcile_orders(self, date: str):
        # Get all marketplace orders for the date
        marketplace_orders = await self.marketplace.get_orders(
            created_after=f"{date}T00:00:00",
            created_before=f"{date}T23:59:59"
        )

        # Get all synced ERP orders for the date
        erp_orders = await self.erp.get_orders_by_date(date)

        # Build lookup maps
        marketplace_ids = {o["id"] for o in marketplace_orders}
        synced_ids = {o["x_external_id"] for o in erp_orders}

        # Find discrepancies
        missing_in_erp = marketplace_ids - synced_ids
        extra_in_erp = synced_ids - marketplace_ids

        if missing_in_erp:
            await self.alert(
                f"MISSING: {len(missing_in_erp)} marketplace orders "
                f"not found in ERP: {missing_in_erp}"
            )
            # Optionally re-queue missing orders
            for order_id in missing_in_erp:
                await self.requeue_order(order_id)

        if extra_in_erp:
            await self.alert(
                f"EXTRA: {len(extra_in_erp)} ERP orders with no "
                f"matching marketplace order (possible test/cancelled)"
            )

I run reconciliation nightly for the previous day's data. It catches roughly 0.1-0.5% of transactions that slip through, depending on the marketplace and volume. That sounds small, but at 1,000 orders per day, 5 missing orders per day compounds into a real problem within a week.

Lessons from 35+ Connectors

1. Every Marketplace Lies About Their API

Documentation says one thing; the API does another. Always build your connector against real API responses, not documentation examples. I maintain a test suite with captured real responses for every marketplace I integrate with.

2. Timestamps Are the Source of All Pain

Marketplaces use different timestamp formats, timezones, and precision. Amazon uses ISO 8601 with timezone offset. Shopify uses ISO 8601 in the shop's timezone. eBay uses their own format. Normalize everything to UTC immediately upon ingestion.

3. Rate Limits Are Not Suggestions

Amazon's SP-API will throttle you aggressively. eBay's Trading API has call limits per day. Shopify's REST API allows 40 requests per app per store per minute. Build rate limiting into your marketplace client from day one, not after you start getting 429 errors in production.

4. Monitor Everything

The metrics that matter:

• Queue depth: How many messages are waiting to be processed
• Processing latency: Time from webhook receipt to ERP record creation
• Error rate: Percentage of messages that fail processing
• Reconciliation delta: Number of discrepancies found daily

If any of these trends upward, you have a problem brewing.

5. Design for the Worst Day, Not the Average Day

Your connector will be tested during Black Friday, Prime Day, or a marketing campaign that nobody told you about. Design your queue capacity, worker count, and timeout values for 10x your average volume. The marginal cost of overprovisioning is tiny compared to the cost of lost orders during a peak event.

Building Your Own Connectors

If you are building marketplace integrations, start with this architecture and adapt it to your specific requirements. The patterns are universal even if the implementation details change between marketplaces and ERPs.

If you want help designing or building a connector, reach out. I have worked with marketplaces across North America, Europe, the Middle East, and South Asia, and I am happy to share what I have learned.

---

Building a marketplace integration and want to avoid the common pitfalls? Get in touch for an architecture review or implementation support.