Commstate Docs

Order Ingest Pipeline

Every order — whether it arrives via Shopify webhook, WooCommerce push, CSV import, or a manual UI entry — enters Commstate through a single canonical pipeline. This ensures uniform deduplication, audit, event publication, and replay across sources.

Architecture

┌────────────────────┐
│ Shopify / Woo      │  POST /api/webhooks/{tenant}/orders/{provider}/{topic}
│ CSV import / UI    │───────────────┐
│ Custom integration │               │
└────────────────────┘               │
                                     ▼
                  ┌────────────────────────────────────┐
                  │ OrderIngestController (sync, <50ms)│
                  │  1. resolve provider via registry  │
                  │  2. verify HMAC signature          │
                  │  3. insert raw payload             │
                  │  4. dispatch NormalizeIngestJob    │
                  │  5. respond 202 Accepted           │
                  └────────────────────┬───────────────┘
                                       │
                                       ▼
                         ┌─────────────────────┐
                         │  order_ingests      │ ← raw payload, replayable
                         │  (pending)          │
                         └──────────┬──────────┘
                                    │
                                    ▼
                   ┌────────────────────────────────┐
                   │ NormalizeIngestJob             │
                   │  (queue: ingest.normalize)     │
                   │  fork on the topic:            │
                   │   order   → processWebhook()   │
                   │   product → processProduct-    │
                   │             Webhook()          │
                   └──────┬──────────────────┬──────┘
                          │                  │
                          ▼                  ▼
       ┌──────────────────────────┐  ┌──────────────────────────┐
       │ OrderIngestService       │  │ ProductIngestService     │
       │ ::upsert()               │  │ ::upsert() / ::delete()  │
       └────────────┬─────────────┘  └────────────┬─────────────┘
                    │                             │
                    ▼                             ▼
      ┌──────────────────────────────┐  ┌────────────────────────┐
      │ orders (canonical)           │  │ products (canonical)   │
      │ OrderEventPublisher fires    │  │ ProductEventPublisher  │
      │ → inventory reservation      │  │ fires                  │
      │ → search indexer             │  └────────────────────────┘
      │ → outbound webhooks          │
      │ → notifications / analytics  │
      └──────────────────────────────┘

Why two stages?

Stage 1 — raw payload capture keeps the provider's retry clock short. Shopify disables webhook endpoints that return 500s; WooCommerce retries aggressively. By ACKing after persisting the raw payload (and before any business logic runs), a normalizer bug can never cause the provider to give up and drop data.

Stage 2 — normalization can fail, retry, and be replayed without any upstream involvement. A fix to the mapping logic followed by UPDATE order_ingests SET status='pending' WHERE status='failed' replays the affected ingests through the new code.

The order_ingests table

Column Purpose
source shopify, woocommerce, manual, csv, api, ...
topic Provider webhook topic (orders/create, order.updated, ...)
external_id The ID in the source system (for dedup)
idempotency_key Provider-provided delivery ID where available
raw_payload (jsonb) Untouched provider payload
headers (jsonb) Request headers (for audit + future signature re-verification)
signature_valid true once HMAC verified
status pending → processing → processed / failed / skipped
order_id FK to the canonical orders row created during normalization
product_id The canonical products row a product delivery produced. No FK — see below
attempts Number of normalize attempts
received_at / processed_at Timestamps for latency tracking

Despite the name, this table is the platform's webhook receipt log, not an orders-only one: product webhooks arrive at the same endpoint, are signed the same way, are deduplicated by the same rules and are retried by the same job. Exactly one of order_id / product_id is populated, according to what the topic turned out to be about.

product_id deliberately carries no foreign key where order_id does. Module migrations run per-module on enable rather than in one ordered pass, so a constraint pointing at products would make the Orders module depend on Products having been migrated first — a dependency it does not declare and should not acquire for a debugging aid.

Deduplication

Two different things arrive looking similar, and the pipeline has to tell them apart: a redelivery (the same event sent twice, absorbed silently) and a second edit (a new change to a resource we already hold, which must be processed).

  • Where the provider sends a delivery id — X-Shopify-Webhook-Id, X-WC-Webhook-Delivery-ID, or Idempotency-Key — that is the answer. Unique index: (source, idempotency_key).
  • Without one, the payload hash is compared against the most recent delivery for the same (source, external_id, topic). Comparing against all history would be wrong: a resource edited A → B → A is entitled to come back to A.

Because topic is part of that key, an order and a product that happen to share an id in their own systems never mask each other.

Duplicate deliveries return 200 OK with status: duplicate without re-processing.

Webhook endpoint

POST /api/webhooks/{tenant}/orders/{provider}/{topic}
  • Unauthenticated — HMAC signature IS the authentication
  • {tenant} — resolved by InitializeTenancyByPath middleware
  • {provider} — looked up in StoreProviderRegistry
  • {topic} — decides which service normalises the delivery (see below)

The orders segment names the ingest endpoint, not the resource. It is historical: this path was in production before products were synced, and every store already has webhooks registered against it. One path is also what makes disconnect cleanup reliable — the registrars find their own hooks by matching that prefix, and a second path for products would be a second rule to keep in step.

Product webhooks

Product events ride the same endpoint, the same signature check, the same receipt row and the same retry policy as orders. The topic is the only thing that distinguishes them, and NormalizeIngestJob forks on it:

provider.isProductWebhookTopic($topic)
  ├─ true  → provider.processProductWebhook()  → ProductIngestService
  └─ false → provider.processWebhook()         → OrderIngestService

processProductWebhook() returns an envelope rather than a bare payload:

[
    'action'      => 'upsert' | 'delete' | 'ignore',
    'external_id' => '7654321098',
    'product'     => [ /* canonical IngestProductDTO shape, upserts only */ ],
]

The envelope exists because a delete carries no product to map — Shopify's products/delete is literally {"id": 123}. Handed to the update path, that payload's absent name, price and stock would be written through as blanks over a real product. delete instead applies products.delete_policy, which archives by default.

Provider Topics registered Also recognised
Shopify products/create, products/update, products/delete —
WooCommerce product.created, product.updated, product.deleted product.restored (pushed by the connector plugin)

Registration lists are config-driven (shopify.product_webhooks, woocommerce.product_webhooks); classification is not. Narrowing the list stops us asking for a delivery, but a store connected while a topic was registered goes on sending it — and a topic we stopped recognising would fall through to the order mapper.

Both providers' processWebhook() refuses product topics outright, so a product payload can never be normalised into an order. That regression is real history: products/update used to sit in the order registration list, and every product edit minted a junk order.

Configuring a Shopify webhook

Set the webhook URL in your Shopify admin (or via the API) to:

https://api.example.com/api/webhooks/<tenant-alias>/orders/shopify/orders-create

The provider's verifyWebhookSignature() method reads the X-Shopify-Hmac-Sha256 header and computes base64(hmac_sha256(raw_body, app_secret)) for comparison.

Implementing a new provider

A provider is any class that implements App\Core\Contracts\WebhookCapableStoreProvider (which extends StoreProviderContract). The easiest path is to extend the shared base:

use App\Core\Stores\BaseStoreProvider;

class MyProvider extends BaseStoreProvider
{
    public function getIdentifier(): string { return 'my-provider'; }
    public function getName(): string        { return 'My Provider'; }

    protected function signatureHeaderName(): string
    {
        return 'x-myprovider-signature';
    }

    protected function webhookSecret(): ?string
    {
        return $this->credential('webhook_secret');
    }

    public function extractExternalId(string $topic, array $payload): ?string
    {
        return $payload['order']['id'] ?? null;
    }

    public function processWebhook(string $topic, array $payload): array
    {
        // Return a canonical IngestOrderDTO-shape array.
        return [
            'external_id' => $this->extractExternalId($topic, $payload),
            'customer'    => [...],
            'items'       => [...],
            // ...
        ];
    }

    // implement the remaining StoreProviderContract methods (fetchOrders, fetchProduct, ...)
}

BaseStoreProvider supplies working defaults for the product-webhook half of the contract — no topics, and ignore for anything that arrives — so a provider that does not sync products need not mention it. One that does overrides three methods:

public function getProductWebhookTopics(): array   // registered on connect
public function isProductWebhookTopic(string $t): bool  // routes the delivery
public function processProductWebhook(string $t, array $payload): array

Register the provider in your module's service provider:

public function boot(\App\Core\Stores\StoreProviderRegistry $registry): void
{
    $registry->register('my-provider', MyProvider::class);
}

Once registered, the webhook path POST /api/webhooks/{tenant}/orders/my-provider/{topic} is live. No changes to the Orders module, no changes to the ingest controller.

Replaying failed ingests

-- Replay everything that failed in the last hour
UPDATE order_ingests
SET status = 'pending', error_message = NULL
WHERE status = 'failed'
  AND received_at > now() - interval '1 hour';

Then dispatch the normalize job for each — a simple command will be added in a follow-up (php artisan orders:replay-ingests).

Queue topology

The normalize job runs on a dedicated ingest.normalize queue with the supervisor-ingest Horizon supervisor (8 workers in production, 2 locally). Failures retry with exponential backoff: 10s → 60s → 300s, then status=failed for manual replay.

Manual orders and CSV imports

Manual orders (created via the admin UI) do not pass through order_ingests — there is no raw payload from an upstream to preserve, and the operator UI is already the source of truth. They call OrderIngestService::ingest($validated, 'manual') directly.

CSV imports (ProcessOrderImportJob) follow the same direct path, storing import status in the separate order_imports table.