Daily Shopify Bulk Sync

Overview

Staff-facing dashboards require fresh sales metrics derived from order headers and order line items. The integration is loosely coupled and asynchronous, implemented via Cron + CF Workflow steps.

Goal: Execute a multi-step CF Workflow that upserts order_header / order_line_item → refreshes product → writes measurement.sales and measurement.performance to R2 for performance tracking and analytics.

API endpoints

Method	Path	Spec	Description
POST	/api/performance-metrics/sync	analytics.yaml	Manual trigger (starts same CF Workflow)
GET	/api/ingest/runs	analytics.yaml	List recent sync telemetry records with status
GET	/orders	order.yaml	List orders (staff, paginated)
GET	/orders/{order_id}	order.yaml	Single order detail
GET	/analytics/commerce/overview	analytics.yaml	Commerce summary KPIs for date range
GET	/analytics/pipelines	analytics.yaml	Pipeline health (SYNC_TELEMETRY stream)

Schema tables

Table	Module	Role
order_header	Order	Write: upserted from Shopify bulk data
order_line_item	Order	Write: upserted from Shopify bulk data
product	Product	Write: refreshed after order upsert

Infrastructure

Resource	Type	Purpose
CF Cron Trigger	Cron	Daily 6am UTC trigger
ingest	Worker (CF Workflow)	Orchestrates multi-step sync via Durable Objects
SYNC_TELEMETRY	Pipeline	Execution audit trail to R2
COMMERCE_TELEMETRY	Pipeline	Order/fulfillment event streaming
R2 bucket	Object storage	JSONL archives, measurement data, telemetry
Durable Object	State store	Watermark tracking

Participants

Actor	Role
CF Cron Trigger	Fires daily at 6am UTC — scheduled trigger
CF Workflow	Orchestrates multi-step sync via Durable Objects
Shopify Admin API	Source of order data
PlanetScale	Database for order records, metrics
R2	Object storage for JSONL archives

Preconditions

• Shopify Admin API access token configured as a Wrangler secret
• PlanetScale reachable via Hyperdrive
• Durable Object contains a valid watermark (hard floor: 128 days back)

Main success scenario (CF Workflow steps)

1. Step: Initiate — Cron fires → triggers CF Workflow run
2. Workflow records telemetry to sync.telemetry on R2 (run_id, process_initiated_at, step=initiated)
3. Step: Request — Workflow initiates Shopify bulkOperationRunQuery with orders/line-items query
4. Step: Poll — Workflow polls Shopify with bulk_operation_id every 16 seconds, up to 128 attempts (~34 min max), until COMPLETED or FAILED
5. Step: Download — On COMPLETED, Workflow downloads JSONL results and stores in R2 (stored file for audit/replay)
6. Step: Parse + Upsert — Workflow parses JSONL in chunks and upserts into order_header and order_line_item
7. Step: Transform — Workflow refreshes product, writes measurement.sales and measurement.performance to R2
8. Step: Finalize — Workflow updates watermark in Durable Object and writes completion to sync.telemetry on R2

sequenceDiagram
  participant Cron as CF Cron
  participant WF as CF Workflow
  participant Shopify as Shopify API
  participant DB as PlanetScale
  participant R2 as R2

  Cron->>WF: trigger workflow run
  WF->>R2: write sync.telemetry(initiated)
  WF->>Shopify: bulkOperationRunQuery
  WF->>Shopify: poll bulk status (16s interval)
  Note over WF,Shopify: retry poll if pending (max 128 attempts)
  alt COMPLETED
    WF->>Shopify: download JSONL
    WF->>R2: put(JSONL archive)
    WF->>DB: upsert order_header (chunked)
    WF->>DB: upsert order_line_item (chunked)
    WF->>DB: refresh product
    WF->>R2: write measurement.sales + measurement.performance
    WF->>R2: write sync.telemetry(completed) + update DO watermark
  else FAILED/RATE-LIMITED
    WF->>WF: retry with backoff
  end

Exception flows

Error condition	Recovery
Bulk op already running	Delay retry — only one concurrent bulk query on API version 2024-04
Shopify rate limit / transient failure	Queue retry + delay (up to 12 hours)
PlanetScale row limit exceeded	Reduce batch size, resume from last committed chunk
JSONL parse error	Mark execution as failed, store error in sync.telemetry on R2

Workflow parameters

Workflow trigger payload

{
  "run_id": "uuid",
  "initiator": "cron|user",
  "requested_at": "ISO8601"
}

Performance SLA

Daily incremental CF Workflow completes in < 15 minutes (async). Poll budget: 128 attempts x 16s = ~34 min hard ceiling. Chunk DB writes to avoid 20s transaction timeout.

Acceptance criteria

• PASS: sync.telemetry on R2 shows completed execution; measurement.performance on R2 updated; sample products show correct 7/30-day sales metrics
• FAIL: Execution ends in failed; watermark regressions; duplicate measurement.sales entries after re-run

Watermark strategy

The incremental sync uses the watermark in the Durable Object to query only orders modified since the last successful run.

• Hard floor: 128 days. If the watermark is older than 128 days (or missing), it resets to NOW() - 128 days. This bounds the maximum data volume per incremental run and prevents unbounded Shopify bulk queries.
• Advance on success only: the watermark advances to MAX(updated_at) from the fetched batch only after all Steps complete. A failed run leaves the watermark unchanged — the next run retries from the same position.

Full backfill

Historical data loads beyond the 128-day window use a separate Durable Object workflow rather than the daily sync path. The backfill workflow:

• Accepts a custom date range (no floor constraint)
• Runs as its own workflow execution with a distinct job_type (FULL_BACKFILL_EXECUTE)
• Uses the same parse/upsert/transform Steps but with larger chunk budgets
• Does not advance the daily watermark — the two are independent
• Can run concurrently with daily sync since Shopify allows one bulk op per app, but the backfill workflow handles the “bulk op already running” error with longer retry delays

Summary (7 buckets)

Bucket	Answer
WHAT	Bulk data acquisition workflow (order data from Shopify)
WHO	CF Workflow, Shopify API, Cron Trigger
WHEN	Daily 6am UTC (cron); on-demand via POST /api/performance-metrics/sync
WHERE	CF Workflow / Durable Objects, R2 bucket, PlanetScale tables
HOW WE KNOW	sync.telemetry on R2, JSONL archives (R2 objects), watermark in Durable Object
HOW IT IS	Execution status: initiated → polling → downloading → upserting → completed/failed
WHY	Keep dashboard data fresh; scheduled data currency