Shopify Kafka Inventory Sync

Event-driven inventory synchronization pipeline that replaces daily batch pulls with near-real-time updates from Shopify webhooks.

Problem

The existing pipeline syncs Shopify inventory data once per day via the Shopify API and Airflow, resulting in stale inventory on customer-facing pages. This project moves to an event-driven push model so inventory updates propagate within seconds.

Documentation

Detailed module documentation with code walkthroughs:

Module	Description
FastAPI Webhook Receiver	HMAC verification, payload validation, Kafka producer
Kafka Event Backbone	KRaft configuration, topic design, message format, CLI debugging
Consumer Service	Poll loop, field mapping, idempotent UPSERT, error handling

Architecture

flowchart LR
    subgraph Shopify
        SW[Shopify Webhook<br/><i>inventory_levels/update</i>]
    end

    subgraph Docker Compose
        subgraph Ingestion
            FA[FastAPI<br/>Webhook Receiver<br/><small>:8000</small>]
        end

        subgraph "Event Backbone"
            K[(Kafka<br/>KRaft Mode<br/><small>:9092 / :9094</small>)]
        end

        subgraph Processing
            C[Consumer Service<br/><small>Python</small>]
        end

        subgraph Storage
            PG[(PostgreSQL 16<br/><small>:5432</small>)]
        end
    end

    subgraph "Local Testing"
        MS[Mock Shopify<br/><small>profile: mock</small>]
    end

    SW -->|HTTP POST +<br/>HMAC-SHA256| FA
    MS -.->|Simulated<br/>Webhooks| FA
    FA -->|Produce| K
    K -->|Consume| C
    C -->|UPSERT| PG

Loading

Event Flow

sequenceDiagram
    participant S as Shopify / Mock
    participant F as FastAPI
    participant K as Kafka
    participant C as Consumer
    participant P as PostgreSQL

    S->>F: POST /webhooks/inventory<br/>(X-Shopify-Hmac-Sha256)
    F->>F: Verify HMAC signature
    F->>F: Validate payload (Pydantic)
    F->>K: Produce to shopify.inventory.update<br/>key: {item_id}_{location_id}
    F-->>S: 200 OK

    K->>C: Poll messages
    C->>C: Deserialize JSON payload
    C->>C: Map fields to DB schema
    C->>P: UPSERT inventory_levels<br/>(idempotent, ordered by updated_at)
    P-->>C: Ack

Loading

Database Schema

erDiagram
    inventory_levels {
        VARCHAR inventory_item_id PK "Shopify inventory item ID"
        VARCHAR warehouse_location_id PK "Shopify location ID"
        VARCHAR variant_id "Internal product hash (nullable)"
        VARCHAR warehouse_name "Resolved from warehouses table"
        INTEGER available "Current stock quantity"
        TIMESTAMPTZ updated_at "From Shopify webhook"
        TIMESTAMPTZ synced_at "Pipeline processing time"
    }

    warehouses {
        VARCHAR warehouse_location_id PK "Shopify location ID"
        VARCHAR warehouse_name "Human-readable name"
    }

    warehouses ||--o{ inventory_levels : "resolves name"

Loading

Schema compatible with typical warehouse inventory tables (e.g. warehouse.stocks_new).

Project Structure

shopify_kafka/
├── fastapi_app/              # Webhook receiver + Kafka producer
│   ├── app/
│   │   ├── config.py         # Pydantic settings
│   │   ├── main.py           # FastAPI app with lifespan
│   │   ├── kafka/
│   │   │   └── producer.py   # KafkaEventProducer
│   │   └── webhooks/
│   │       ├── router.py     # POST /webhooks/inventory
│   │       ├── schemas.py    # InventoryLevelUpdate model
│   │       └── verification.py  # HMAC-SHA256 verification
│   ├── tests/                # 10 unit tests
│   ├── Dockerfile
│   └── requirements.txt
│
├── consumer/                 # Kafka consumer + Postgres writer
│   ├── app/
│   │   ├── config.py         # Pydantic settings (Kafka + Postgres DSN)
│   │   ├── consumer.py       # InventoryConsumer
│   │   ├── main.py           # Entry point
│   │   └── db/
│   │       └── repository.py # Idempotent UPSERT logic
│   ├── tests/                # 4 unit tests
│   ├── Dockerfile
│   └── requirements.txt
│
├── mock_shopify/             # Simulated Shopify webhooks for local testing
│   ├── app/
│   │   ├── config.py         # Interval, target URL settings
│   │   ├── generator.py      # Payload + HMAC signing
│   │   └── main.py           # Continuous webhook sender
│   ├── Dockerfile
│   └── requirements.txt
│
├── postgres/
│   └── init.sql              # Schema + seed data
│
├── docker-compose.yml        # All services orchestrated
├── .env.example              # Environment variable template
└── GUIDE.md                  # Architecture background

Quick Start

Prerequisites

• Docker & Docker Compose

1. Configure environment

cp .env.example .env

Before publishing: Never commit .env (it is in .gitignore). The repo uses only sample data and placeholder secrets for local development.

2. Start the pipeline

docker compose up --build -d

This starts four services: Kafka, PostgreSQL, FastAPI, and Consumer.

3. Verify services are healthy

docker compose ps

All services should show Up (Kafka and Postgres should show healthy).

4. Start mock Shopify (generates test events)

docker compose --profile mock up -d mock_shopify --build

5. Watch the pipeline in action

# Mock sending events
docker compose logs -f mock_shopify

# FastAPI receiving + producing to Kafka
docker compose logs -f fastapi_app

# Consumer upserting to Postgres
docker compose logs -f consumer

6. Query inventory in Postgres

docker compose exec postgres psql -U inventory_user -d inventory -c \
  "SELECT il.inventory_item_id, w.warehouse_name, il.available, il.updated_at
   FROM inventory_levels il
   LEFT JOIN warehouses w ON il.warehouse_location_id = w.warehouse_location_id
   ORDER BY il.synced_at DESC;"

7. Tear down

docker compose --profile mock down -v

Services

graph TD
    subgraph "Docker Compose Services"
        kafka["<b>kafka</b><br/>apache/kafka:latest<br/>KRaft mode, no Zookeeper<br/>Ports: 9092 (internal), 9094 (external)"]
        postgres["<b>postgres</b><br/>postgres:16<br/>inventory_levels + warehouses<br/>Port: 5432"]
        fastapi["<b>fastapi_app</b><br/>Python / FastAPI<br/>Webhook receiver + Kafka producer<br/>Port: 8000"]
        consumer["<b>consumer</b><br/>Python / confluent-kafka<br/>Kafka consumer + Postgres writer"]
        mock["<b>mock_shopify</b><br/>Python<br/>Simulated Shopify webhooks<br/><i>Profile: mock</i>"]
    end

    kafka -.->|healthcheck| kafka
    postgres -.->|healthcheck| postgres
    fastapi -->|depends_on healthy| kafka
    consumer -->|depends_on healthy| kafka
    consumer -->|depends_on healthy| postgres
    mock -->|depends_on started| fastapi

Loading

Service	Image	Port	Depends On
kafka	apache/kafka:latest	9094 (external)	--
postgres	postgres:16	5432	--
fastapi_app	Built from ./fastapi_app	8000	kafka (healthy)
consumer	Built from ./consumer	--	kafka (healthy), postgres (healthy)
mock_shopify	Built from ./mock_shopify	--	fastapi_app (started)

Testing

Unit Tests (on host)

# FastAPI tests (10 tests)
cd fastapi_app && pip install -r requirements.txt && python -m pytest tests/ -v

# Consumer tests (4 tests)
cd consumer && pip install -r requirements.txt && python -m pytest tests/ -v

Note: confluent-kafka requires librdkafka and psycopg2-binary requires libpq on the host. On macOS: brew install librdkafka libpq.

End-to-End Test

Start the full pipeline with the mock profile and verify rows appear in PostgreSQL:

docker compose --profile mock up --build -d
sleep 15
docker compose exec postgres psql -U inventory_user -d inventory \
  -c "SELECT COUNT(*) FROM inventory_levels;"

Configuration

All configuration is via environment variables (.env file):

Variable	Default	Description
SHOPIFY_WEBHOOK_SECRET	test-secret-for-local-dev	HMAC shared secret for webhook verification
KAFKA_BOOTSTRAP_SERVERS	kafka:9092	Kafka broker address
KAFKA_TOPIC	shopify.inventory.update	Kafka topic for inventory events
POSTGRES_HOST	postgres	PostgreSQL host
POSTGRES_PORT	5432	PostgreSQL port
POSTGRES_DB	inventory	PostgreSQL database name
POSTGRES_USER	inventory_user	PostgreSQL username
POSTGRES_PASSWORD	inventory_pass	PostgreSQL password
MOCK_SHOPIFY_INTERVAL_SECONDS	5	Seconds between mock webhook sends
MOCK_SHOPIFY_TARGET_URL	http://fastapi_app:8000/webhooks/inventory	Target endpoint for mock webhooks

Key Design Decisions

• FastAPI does not write to Postgres -- keeps the webhook receiver thin (accept, validate, publish). Kafka provides durability and retry.
• Idempotent UPSERTs -- ON CONFLICT ... WHERE updated_at < EXCLUDED.updated_at ensures out-of-order messages don't overwrite newer data.
• Kafka message key -- {inventory_item_id}_{location_id} guarantees ordering per inventory item within a partition.
• Mock Shopify as a Docker profile -- isolated behind --profile mock so it doesn't run in production-like deployments.
• Schema alignment -- inventory_levels columns match common warehouse inventory naming conventions for easy integration.

Roadmap

• Real Shopify webhook subscription -- requires Shopify Partner app credentials + public endpoint (ngrok for dev)
• S3 Data Lake -- daily historical snapshots from Kafka or Postgres
• Observability -- structured logging, Prometheus metrics, dead letter queue for failed messages
• variant_id enrichment -- lookup service to map inventory_item_id to internal variant_id hash
• Kubernetes deployment -- Helm charts, StatefulSets for Kafka and Postgres
• Horizontal scaling -- multiple Kafka partitions + consumer group scaling
• Schema registry -- Avro/Protobuf schemas for Kafka messages (currently plain JSON)