Event processing

subscriber.management.commands.run_subscriber owns the message lifecycle. It accepts either a wrapped event envelope or a raw Shopify JSON object accompanied by Pub/Sub attributes.

Accepted message shapes

Wrapped envelope

{
  "id": "webhook-uuid",
  "source": "shopify",
  "type": "orders.create",
  "shop_domain": "store.myshopify.com",
  "timestamp": "2026-06-22T02:46:46Z",
  "payload": {
    "id": 1001,
    "name": "#1001"
  }
}

Raw body with Pub/Sub attributes

Message data can be the Shopify body itself:

{
  "id": 1001,
  "name": "#1001"
}

The publisher attaches attributes such as X-Shopify-Topic, X-Shopify-Shop-Domain, and X-Shopify-Webhook-Id. When Shopify headers are present and the JSON does not already contain payload, the subscriber wraps the original object as payload before dispatch.

Header extraction and precedence

Envelope headers and Pub/Sub attributes are merged into headers. Attributes overwrite same-key envelope header entries. Lookup is case-insensitive.

Output field	First non-empty value wins
type	envelope.type, attribute type, X-Shopify-Topic, header topic
shop_domain	envelope.shop_domain, attribute shop_domain, X-Shopify-Shop-Domain
id / webhook_id	event_id, webhook_id, matching attributes, X-Shopify-Webhook-Id, then existing id
source	envelope.source, attribute source, otherwise inferred as shopify when Shopify headers exist
pubsub_message_id	Pub/Sub message identifier when present

Topics are normalized from slash form to dot form before routing: orders/create becomes orders.create. The canonical slash topic is restored inside the parsed event for storage.

Dispatch pipeline

Router

subscriber/router.py builds one mapping entry for every allowlisted event type. The resource prefix selects one of 12 handler instances. Unknown sources and non-allowlisted topics return no handler.

Handler

Each module owns a small handler class, for example ShopifyOrdersHandler or ShopifyProductsHandler. The shared ShopifyEventHandler extracts type and payload, invokes the service, logs the resolved resource, and delegates persistence.

The customer handler performs an additional guard for the five supported customer topics before entering the shared path.

Service

Each service declares one resource_type. ShopifyResourceService.parse() calls the shared parser and rejects a parsed event whose resource type does not match the service. persist() delegates to ShopifyEventStorage.

This thin service layer establishes a module boundary without duplicating the common SQL implementation.

Parsing behavior

All 39 topics receive structural parsing:

• Normalize slash or dot topic notation.
• Verify the topic is in SHOPIFY_TRACKED_EVENT_TYPES.
• Require payload to be a JSON object.
• Split the topic into resource_type and operation.
• Resolve the destination table from RESOURCE_TABLE_BY_TYPE.
• Extract a resource ID from id, admin_graphql_api_id, graphql_id, or order_id.
• Set is_deleted only when the operation name is exactly delete.

Customer create, update, and delete events additionally use the existing customer dataclass parsers and convert the result to a JSON dictionary. For all other tracked events, including customer enable/disable, parsed_payload currently falls back to the original payload.

Parser coverage

Separate module services do not yet mean field-specific transformation for every module. Orders, products, refunds, returns, and the other non-customer modules currently store normalized metadata plus the original JSON. Future typed parsers can be added behind the existing service classes without changing routing or storage.

Event identity and timestamps

Storage uses the first available event identifier from id, event_id, or webhook_id. If none exists, it computes a deterministic SHA-256 hash over source, type, shop_domain, and payload.

received_at is parsed from received_at or timestamp. A timestamp without timezone information is treated as UTC. If no valid value is provided, the backend's current time is used.

Acknowledgement outcomes

Outcome	Pub/Sub action	Durable failure row
Handler and storage complete	ack	No
JSON, UTF-8, payload, parser, or storage error; failure row saved	ack	Yes, pending
Event processing fails and failure row cannot be saved	nack	No; Pub/Sub must redeliver
No handler for source/type	ack after dispatch returns	No
No matching or duplicate Shopify connection	ack after dispatch returns	No

The last two outcomes are intentional current behavior, but they are observability gaps because the event is not replayable from SubscriberFailedEvent.

Transaction boundary

The target event upsert and optional resource snapshot upsert run through one psycopg connection context. PostgreSQL commits both together on a successful context exit and rolls the transaction back if either statement raises.

Schema checks run on the same connection before the inserts. They use CREATE TABLE IF NOT EXISTS and CREATE INDEX IF NOT EXISTS, making normal event processing self-healing for missing tables while the explicit provisioning command remains the recommended onboarding path.