Commerce Sync Platform

Commerce integration platform for synchronizing orders with downstream systems. It combines webhook ingestion, reconciliation polling, vendor adapters, canonical mapping, Kafka workers, layered idempotency, durable persistence, DLQ replay, rate limiting, and OpenTelemetry observability.

The current release delivers Shopify synchronization and a SuiteTalk REST NetSuite sales-order adapter through the same connector contract.

Run the complete local product demonstration with:

npm run demo

Product Status

The current product baseline implements and tests the complete synchronization core and its principal failure-handling paths. Administrative operations use API-key protection, and the roadmap evolves the platform with identity/RBAC, tenant administration, managed secrets, transactional outbox delivery, and the NetSuite adapter.

Capability	Current state
Shopify webhook ingestion	Implemented with HMAC verification and duplicate suppression
Shopify polling reconciliation	Implemented with cursor/watermark persistence
Kafka worker processing	Implemented
Idempotent Postgres upsert	Implemented with stale-update protection
Durable DLQ and replay	Implemented
Observability stack	Provisioned Grafana dashboard, Prometheus alerts and SLO targets
Versioned API contract	OpenAPI 3 at `/docs-json`; Swagger UI at `/docs`
Request safety	DTO validation, payload limits, Helmet and consistent errors
Request correlation	Propagated across HTTP, Kafka, logs, traces and DLQ
Tenant integration accounts	PostgreSQL-backed, tenant-scoped, external secret references only
Webhook recovery	Durable leasing, exponential retry and multi-replica-safe claiming
Asynchronous backfills	Kafka-dispatched jobs with leases, crash recovery, tenant isolation and status history
DLQ replay governance	Idempotency keys, actor audit trail and replay-storm protection
NetSuite adapter	SuiteTalk REST pagination, canonical mapping, reads and idempotent note updates
Authentication and tenant administration	Service identity, API key and tenant policy implemented; OAuth/OIDC and full RBAC are next
HA and disaster-recovery automation	Target architecture and operating procedures defined

Documentation

Area	Documents
Product	Product brief · Roadmap and status · Product vision · Non-functional requirements · Release readiness · Compatibility
Architecture	System context · HLD · LLD · Deployment
Infrastructure	AWS HA Terraform · Kafka scaling
Decisions	Architecture Decision Records
Operations	Local development · Runbook · Troubleshooting · Disaster recovery · SLOs
Guided product tour	8–12 minute walkthrough

System at a Glance

flowchart TB
  Merchant["Merchant / Operator"]
  Shopify["Shopify-compatible API"]

  subgraph Platform["Commerce Sync Platform"]
    API["API service<br/>webhooks + operations"]
    Scheduler["Scheduler service<br/>reconciliation polling"]
    Worker["Worker service<br/>event processing"]
  end

  Kafka[("Kafka")]
  Redis[("Redis")]
  Postgres[("PostgreSQL")]
  Telemetry["OTel / Prometheus / Jaeger"]

  Shopify -->|signed webhooks| API
  Scheduler -->|poll changed orders| Shopify
  Merchant -->|sync and replay operations| API
  API --> Kafka
  Kafka --> Worker
  API --> Redis
  Scheduler --> Redis
  Worker --> Redis
  API --> Postgres
  Scheduler --> Postgres
  Worker --> Postgres
  Platform --> Telemetry

Why This Exists

Commerce and ERP sync systems fail in predictable ways: duplicate webhooks, out-of-order updates, API rate limits, vendor outages, partial polling failures, worker retries, and data drift. This repo demonstrates the backend patterns used to make those failures safe and debuggable.

Architecture

flowchart LR
  Shopify["Shopify / Commerce<br/>webhooks + API"]

  subgraph API["api-service"]
    Webhook["WebhookController<br/>HMAC + inbox + idempotency"]
    Health["Health / Ready"]
  end

  subgraph Scheduler["scheduler-service"]
    Cron["SyncScheduler<br/>10 min cron"]
    Lock["Redis distributed lock<br/>tenant + vendor"]
  end

  subgraph Worker["worker-service"]
    Consumer["SyncWorker<br/>Kafka consumer group"]
  end

  subgraph Core["sync core"]
    Connector["ShopifyConnector"]
    Mapper["Shopify mapper<br/>vendor -> canonical Order"]
    Sync["SyncService<br/>retry + bounded concurrency"]
    Repo["OrderRepository<br/>idempotent upsert"]
    DLQ["DeadLetterQueue<br/>persist + replay"]
  end

  Kafka["Kafka<br/>commerce.events / commerce.dlq"]
  Redis["Redis<br/>SET NX idempotency<br/>token bucket<br/>scheduler lock"]
  PG["Postgres<br/>orders<br/>sync_state<br/>sync_runs<br/>webhook_events<br/>dead_letter_events"]
  OTel["OpenTelemetry Collector"]
  Jaeger["Jaeger"]
  Prom["Prometheus / Grafana"]

  Shopify -->|signed webhook| Webhook
  Webhook -->|record received/published/failed| PG
  Webhook -->|claim event id| Redis
  Webhook -->|publish tenant-keyed event| Kafka
  Kafka -->|consume commerce.events| Consumer
  Consumer --> Mapper
  Consumer --> Repo
  Consumer -->|failure| DLQ
  DLQ -->|publish replay| Kafka

  Cron --> Lock
  Lock --> Sync
  Sync --> Connector
  Connector -->|poll updated_since watermark| Shopify
  Sync --> Mapper
  Sync --> Repo
  Sync -->|save cursor + run history| PG
  Sync -->|row failure| DLQ

  Repo --> PG
  DLQ --> PG
  API --> OTel
  Scheduler --> OTel
  Worker --> OTel
  OTel --> Jaeger
  OTel --> Prom

Runtime Flow

sequenceDiagram
  participant Commerce as Shopify
  participant API as api-service
  participant Redis
  participant Kafka
  participant Worker as worker-service
  participant PG as Postgres
  participant DLQ as DLQ

  Commerce->>API: POST /api/v1/webhooks/shopify
  API->>API: verify HMAC signature
  API->>PG: insert webhook_events(received)
  API->>Redis: claim event id with SET NX PX
  alt duplicate event
    API-->>Commerce: 200 duplicate
  else new event
    API->>Kafka: publish commerce.events keyed by tenantId
    API->>PG: mark webhook_events(published)
    API-->>Commerce: 200 accepted
  end

  Kafka->>Worker: consume event
  Worker->>Worker: map vendor payload to Order
  Worker->>PG: INSERT ... ON CONFLICT ... WHERE remote_updated_at is newer
  alt success
    Worker->>PG: order inserted/updated/skipped
  else permanent failure
    Worker->>DLQ: persist dead_letter_events
    Worker->>Kafka: publish commerce.dlq
  end

What It Demonstrates

API, worker, and scheduler split from the same codebase.
HMAC-validated webhook ingestion.
Durable webhook inbox table.
Redis idempotency for duplicate webhook suppression.
Kafka topic handoff from ingress to worker.
Per-tenant Kafka keying for ordering.
Scheduled polling with durable sync_state watermark.
Redis distributed scheduler lock scoped by tenant/vendor.
Shopify connector and explicit mapper to canonical Order.
Postgres idempotent order upsert using (tenant_id, vendor, external_id).
Out-of-order protection via remote_updated_at.
Redis token-bucket rate limiting.
Bounded retries and concurrency.
Durable DLQ records with tenant, vendor, operation, category, retry count, correlation ID, and payload.
DLQ replay endpoint back to commerce.events.
OpenTelemetry traces and metrics through OTel Collector, Jaeger, Prometheus, and Grafana.
Docker Compose stack for Postgres, Redis, Kafka, Kafka UI, Jaeger, Prometheus, and Grafana.

Main Tables

Table	Purpose
`orders`	Latest canonical order state. Natural key: `(tenant_id, vendor, external_id)`.
`sync_state`	Durable polling watermark scoped by tenant, vendor, resource type, and integration account.
`sync_runs`	Sync run history, counters, status, and error message.
`webhook_events`	Webhook inbox for received, published, and failed webhook ingress.
`dead_letter_events`	Durable DLQ records for failed polling/worker events.

Schema is in drizzle/0000_initial_sync_tables.sql. Partitioning and retention examples are in drizzle/examples/partitioning_retention.sql.

Project Layout

src/
  main.ts                       api-service entrypoint
  worker.ts                     worker-service entrypoint
  scheduler.ts                  scheduler-service entrypoint
  commerce/                     vendor connector interface + facade
  integrations/shopify/         implemented demo connector and mapper
  integrations/netsuite/        explicit stub
  webhook/                      HMAC webhook ingress + inbox repository
  sync/                         scheduler, worker, sync engine, DLQ, replay, locks
  persistence/                  Postgres-backed repositories
  redis/                        idempotency, rate limiter, Redis client
  kafka/                        KafkaJS wrapper
  db/                           Drizzle schema
  metrics/                      OpenTelemetry metrics
  health/                       health/readiness endpoints
test/                           unit, integration-style, and e2e tests

Run Locally

Install and build:

npm install
npm run build

Run the infrastructure and all three service roles:

docker compose up --build

Run the Shopify mock on the host:

npm run mock:shopify

Useful URLs:

Tool	URL
API	`http://localhost:3000`
Swagger UI	`http://localhost:3000/docs`
OpenAPI JSON	`http://localhost:3000/docs-json`
Kafka UI	`http://localhost:8080`
Jaeger	`http://localhost:16686`
Prometheus	`http://localhost:9090`
Grafana	`http://localhost:3001`

Run roles directly during development:

npm run start:dev:api
npm run start:dev:worker
npm run start:dev:scheduler

Docker Compose runs the same image with three commands:

api-service       node dist/main
worker-service    node dist/worker
scheduler-service node dist/scheduler

Kafka bootstrap addresses:

inside Docker: kafka:29092
from host:     localhost:9092

Sample Webhook

The local compose secret is secret.

BODY='{"tenant_id":"00000000-0000-0000-0000-000000000001","id":1001,"email":"ana@example.com","customer":{"first_name":"Ana","last_name":"Lopez"},"financial_status":"paid","fulfillment_status":"unfulfilled","currency":"CAD","total_price":"129.95","line_items":[{"id":501,"quantity":2}],"updated_at":"2026-05-01T12:00:00.000Z"}'
SIG=$(printf '%s' "$BODY" | openssl dgst -sha256 -hmac 'secret' -binary | openssl base64 -A)

curl -X POST http://localhost:3000/api/v1/webhooks/shopify \
  -H "content-type: application/json" \
  -H "x-shopify-webhook-id: evt_123" \
  -H "x-shopify-hmac-sha256: $SIG" \
  -d "$BODY"

Expected first response:

{ "status": "accepted" }

Expected duplicate response with the same event ID:

{ "status": "duplicate" }

Useful Commands

# health
curl http://localhost:3000/api/v1/health
curl http://localhost:3000/api/v1/health/ready

# registered vendors
curl http://localhost:3000/api/v1/commerce/vendors \
  -H 'x-admin-api-key: local-demo-admin-key' \
  -H 'x-principal-id: service:local-operator'

# manual polling sync
curl -X POST http://localhost:3000/api/v1/sync/orders/start \
  -H 'x-admin-api-key: local-demo-admin-key' \
  -H 'x-principal-id: service:local-operator' \
  -H 'x-principal-roles: sync-operator' \
  -H 'x-principal-tenant-ids: 00000000-0000-0000-0000-000000000001' \
  -H 'content-type: application/json' \
  -d '{"tenantId":"00000000-0000-0000-0000-000000000001"}'

# DLQ status
curl http://localhost:3000/api/v1/sync/dlq \
  -H 'x-admin-api-key: local-demo-admin-key' \
  -H 'x-principal-id: service:local-operator' \
  -H 'x-principal-tenant-ids: 00000000-0000-0000-0000-000000000001'

# replay a DLQ event
curl -X POST http://localhost:3000/api/v1/sync/dlq/webhook:shopify:00000000-0000-0000-0000-000000000001:event-1/replay \
  -H 'x-admin-api-key: local-demo-admin-key' \
  -H 'x-principal-id: service:local-operator' \
  -H 'x-principal-roles: sync-operator' \
  -H 'x-principal-tenant-ids: 00000000-0000-0000-0000-000000000001' \
  -H 'content-type: application/json' \
  -d '{"removeAfterPublish":false,"idempotencyKey":"incident-2026-07-25-order-1001-attempt-1"}'

# controlled historical backfill; the incremental cursor is never changed
curl -X POST http://localhost:3000/api/v1/sync/backfills \
  -H 'content-type: application/json' \
  -H 'x-admin-api-key: local-demo-admin-key' \
  -H 'x-principal-id: service:local-operator' \
  -H 'x-principal-roles: sync-operator' \
  -H 'x-principal-tenant-ids: 00000000-0000-0000-0000-000000000001' \
  -d '{"tenantId":"00000000-0000-0000-0000-000000000001","fromDate":"2026-01-01T00:00:00Z","toDate":"2026-01-31T23:59:59Z"}'

# inspect the queued/running/terminal state returned by the worker
curl 'http://localhost:3000/api/v1/sync/backfills/JOB_ID?tenantId=00000000-0000-0000-0000-000000000001' \
  -H 'x-admin-api-key: local-demo-admin-key' \
  -H 'x-principal-id: service:local-operator' \
  -H 'x-principal-roles: integration-admin' \
  -H 'x-principal-tenant-ids: 00000000-0000-0000-0000-000000000001'

# recover durable inbox events (platform operator)
curl -X POST 'http://localhost:3000/api/v1/operations/webhook-recovery/run?limit=25' \
  -H 'x-admin-api-key: local-demo-admin-key' \
  -H 'x-principal-id: service:platform-operator' \
  -H 'x-principal-roles: platform-admin'

# register a tenant integration account without storing a plaintext secret
curl -X POST http://localhost:3000/api/v1/tenants/00000000-0000-0000-0000-000000000001/integration-accounts \
  -H 'content-type: application/json' \
  -H 'x-admin-api-key: local-demo-admin-key' \
  -H 'x-principal-id: service:local-operator' \
  -H 'x-principal-tenant-ids: 00000000-0000-0000-0000-000000000001' \
  -d '{"vendor":"shopify","displayName":"Primary store","secretReference":"env:SHOPIFY_ACCESS_TOKEN"}'

# Kafka topics
docker compose exec kafka kafka-topics \
  --bootstrap-server localhost:29092 \
  --list

# consumer lag
docker compose exec kafka kafka-consumer-groups \
  --bootstrap-server localhost:29092 \
  --describe \
  --group sync-worker-group

Asynchronous backfill lifecycle

sequenceDiagram
  actor Operator
  participant API
  participant DB as PostgreSQL
  participant Kafka
  participant Worker
  participant Shopify

  Operator->>API: POST /sync/backfills
  API->>DB: insert job (queued)
  API->>Kafka: publish jobId
  API-->>Operator: 202-style queued job
  Kafka->>Worker: deliver job
  Worker->>DB: atomic claim + lease
  Worker->>Shopify: read bounded order window
  Worker->>DB: upsert orders; preserve incremental cursor
  Worker->>DB: succeeded / failed + report
  Operator->>API: GET /sync/backfills/:jobId
  API->>DB: tenant-scoped lookup
  API-->>Operator: current state and execution evidence
  Note over Worker,DB: Polling reclaims queued or expired leases after crashes

Tests

The complete release gate runs serially to remain usable on a developer laptop:

npm run release:check

The reproducible reference microbenchmark is:

BENCHMARK_SAMPLES=5000 npm run benchmark

On a Docker-enabled release host, validate backup restoration with:

npm run restore:drill

CI independently applies every migration to PostgreSQL 16, verifies forced RLS and the outbox, validates both Compose models and Terraform, builds the image, and blocks high/critical fixable container vulnerabilities.

npm run lint
npm test
npm run test:e2e
npm run build
docker compose config --quiet

Current tests cover mapper validation, order upsert semantics, webhook HMAC/idempotency/inbox behavior, worker processing, DLQ replay, distributed lock behavior, sync cursor/watermark behavior, and a webhook-to-worker full sync flow that proves older updates are skipped.

Operational Debugging

When a customer says an order is missing:

Search webhook_events by vendor event ID or tenant/time.
Check Kafka commerce.events and consumer lag for sync-worker-group.
Check dead_letter_events for the tenant, vendor, correlation ID, or error category.
Check orders by (tenant_id, vendor, external_id).
Open Jaeger and inspect the trace for webhook receipt, Kafka publish/consume, Redis, and Postgres spans.
Check Prometheus/Grafana for sync_records_processed_total, sync_duration_seconds, sync_dlq_total, vendor latency, and vendor errors.

Scalability Notes

API services are stateless; scale horizontally behind shared Redis, Kafka, and Postgres.
Workers scale through Kafka consumer groups. Parallelism is bounded by topic partitions and hot tenants.
Scheduler uses a Redis lock per tenant/vendor to avoid duplicate polling across replicas.
Vendor API pressure is controlled by token-bucket rate limiting and bounded concurrency.
Postgres correctness comes from natural keys and conditional upserts.
Large history tables should not grow forever online. Use tenant/time indexes, mandatory date ranges, partitioning, retention, and cold archive.
A very large tenant may need separate topics, worker pools, or database sharding later.

Product Roadmap Priorities

Complete the NetSuite connector against a sandbox account.
Persist tenant integration configuration and connect managed secrets.
Extend API-key protection with identity, RBAC, and administrative workflows.
Provision Kafka topics, partitions, ACLs, retention, and schemas declaratively.
Provision the Grafana dashboards and automate retention/partition jobs.

Interview Talking Points

Webhooks reduce latency; polling catches missed or delayed vendor events.
Idempotency is layered: Redis suppresses duplicate webhook work, Postgres protects final state.
Workers are separate from API so webhook acknowledgements stay fast.
Connectors hide vendor auth, pagination, rate limits, and response shape.
remote_updated_at prevents older out-of-order events from overwriting newer data.
DLQ replay is explicit and auditable.
Sync lag debugging starts with traces, then consumer lag, DLQ growth, vendor latency, and DB write behavior.

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
docs		docs
drizzle		drizzle
infra/terraform		infra/terraform
mocks		mocks
observability		observability
scripts		scripts
src		src
test		test
.dockerignore		.dockerignore
.env.example		.env.example
.gitignore		.gitignore
CHANGELOG.md		CHANGELOG.md
CONTRIBUTING.md		CONTRIBUTING.md
Dockerfile		Dockerfile
README.md		README.md
SECURITY.md		SECURITY.md
docker-compose.scale.yml		docker-compose.scale.yml
docker-compose.yml		docker-compose.yml
drizzle.config.ts		drizzle.config.ts
eslint.config.mjs		eslint.config.mjs
nest-cli.json		nest-cli.json
otel-collector-config.yaml		otel-collector-config.yaml
package-lock.json		package-lock.json
package.json		package.json
prometheus.yml		prometheus.yml
tsconfig.build.json		tsconfig.build.json
tsconfig.json		tsconfig.json

Navigation Menu

Repository files navigation

Commerce Sync Platform

Product Status

Documentation

System at a Glance

Why This Exists

Architecture

Runtime Flow

What It Demonstrates

Main Tables

Project Layout

Run Locally

Sample Webhook

Useful Commands

Asynchronous backfill lifecycle

Tests

Operational Debugging

Scalability Notes

Product Roadmap Priorities

Interview Talking Points

About

Releases

Packages

Contributors

Languages

Latest commit

History

Folders and files

Repository files navigation

Commerce Sync Platform

Product Status

Documentation

System at a Glance

Why This Exists

Architecture

Runtime Flow

What It Demonstrates

Main Tables

Project Layout

Run Locally

Sample Webhook

Useful Commands

Asynchronous backfill lifecycle

Tests

Operational Debugging

Scalability Notes

Product Roadmap Priorities

Interview Talking Points

About

Topics

Resources

Contributing

Security policy

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages