ShipBob BigQuery Integration

Status: Planned (Jan 28, 2026)

This document describes the implementation plan for syncing ShipBob fulfillment data into BigQuery via the eli-kpi service.

Table of Contents

• Overview
• Architecture
• ShipBob API
• BigQuery Schema
• Sync Strategies
• Implementation Details
• Cloud Scheduler
• Verification

Overview

ShipBob (now branded as Jetpack) is Eli Health's fulfillment partner. This integration pulls order, product, inventory, return, location, and channel data into BigQuery for analytics and cross-referencing with Shopify, Loopwork, and other datasets.

What Data We Sync

Table	Source Endpoint	Records	Sync Strategy
orders	GET /order	~3,500+	Rolling window (7-day MERGE)
products	GET /product	~36	Full refresh
inventory	GET /inventory	~21	Full refresh
returns	GET /return	~1	Full refresh
locations	GET /location	~20	Full refresh
channels	GET /channel	5	Full refresh

Architecture

Data Pipeline Context

This integration adds ShipBob as a new data source alongside existing syncs:

ShipBob API

Authentication

• Method: Personal Access Token (PAT) via Bearer header
• Token: Stored as SHIPBOB_API_TOKEN environment variable
• Lifetime: Never expires
• Header: Authorization: Bearer <token>

Rate Limits

• 150 requests per minute (sliding window)
• Response headers: x-retry-after, x-remaining-calls
• On HTTP 429: Wait x-retry-after seconds, then retry
• Implementation: 400ms delay between requests (safe margin)

Pagination

• Type: Offset-based
• Parameters: Page (1-indexed) + Limit (max 250)
• End condition: Empty array returned
• Example: GET /order?Page=1&Limit=250

Endpoints

Endpoint	Method	Pagination	Incremental Filter
/order	GET	Page/Limit (250)	LastUpdateStartDate, LastUpdateEndDate
/product	GET	Page/Limit (250)	None
/inventory	GET	Page/Limit (250)	None
/return	GET	Page/Limit (250)	None
/location	GET	None (returns all)	None
/channel	GET	Cursor-based	None

Key Data Shapes

Order (nested, includes shipments):

{
  "id": 331899137,
  "created_date": "2026-01-28T20:23:15Z",
  "purchase_date": "2026-01-28T19:52:59Z",
  "reference_id": "6095829762093",
  "order_number": "9965",
  "status": "ImportReview",
  "type": "DTC",
  "channel": { "id": 433239, "name": "elihealth" },
  "shipping_method": "USA Standard",
  "recipient": {
    "name": "...",
    "address": { "address1", "city", "state", "country", "zip_code" },
    "email": "...",
    "phone_number": null
  },
  "products": [{ "id", "reference_id", "quantity", "sku", "unit_price" }],
  "shipments": [{
    "id", "status", "tracking": { "tracking_number", "carrier" },
    "estimated_fulfillment_date", "actual_fulfillment_date", "delivery_date",
    "invoice_amount", "ship_option"
  }],
  "tags": []
}

Product:

{
  "id": 1377044214,
  "reference_id": "44195829579821",
  "sku": "VIP-KIT-CL-2",
  "name": "Pack of 2 cortisol tests",
  "unit_price": 50.0,
  "total_fulfillable_quantity": 0,
  "total_onhand_quantity": 0,
  "total_committed_quantity": 0,
  "fulfillable_inventory_items": [{ "id", "name", "quantity" }]
}

Inventory (with lot tracking — Cortisol Pack has is_lot: true):

{
  "id": 21743786,
  "name": "Cortisol Pack",
  "is_active": true,
  "is_lot": true,
  "total_fulfillable_quantity": 10926,
  "total_onhand_quantity": 10975,
  "total_committed_quantity": 49,
  "total_sellable_quantity": 10926,
  "total_awaiting_quantity": 949,
  "total_exception_quantity": 0,
  "total_backordered_quantity": 0,
  "fulfillable_quantity_by_fulfillment_center": [
    { "id": 178, "name": "Elwood (IL)", "fulfillable_quantity": 10798, "onhand_quantity": 10839 },
    { "id": 146, "name": "Brampton (Ontario) 2", "fulfillable_quantity": 128, "onhand_quantity": 136 }
  ],
  "fulfillable_quantity_by_lot": [
    {
      "lot_number": "CB0016",
      "expiration_date": "2027-11-17T00:00:00",
      "fulfillable_quantity": 2296,
      "onhand_quantity": 2296,
      "committed_quantity": 0,
      "awaiting_quantity": 0,
      "fulfillable_quantity_by_fulfillment_center": [
        { "id": 178, "name": "Elwood (IL)", "fulfillable_quantity": 2296, "onhand_quantity": 2296 }
      ]
    },
    {
      "lot_number": "CB0015",
      "expiration_date": "2027-10-31T00:00:00",
      "fulfillable_quantity": 4450,
      "onhand_quantity": 4450,
      "fulfillable_quantity_by_fulfillment_center": [
        { "id": 178, "name": "Elwood (IL)", "fulfillable_quantity": 4450 }
      ]
    },
    {
      "lot_number": "CB0012",
      "expiration_date": "2027-08-12T00:00:00",
      "fulfillable_quantity": 128,
      "onhand_quantity": 136,
      "fulfillable_quantity_by_fulfillment_center": [
        { "id": 146, "name": "Brampton (Ontario) 2", "fulfillable_quantity": 128 }
      ]
    }
  ]
}

Return:

{
  "id": 3090197,
  "status": "Completed",
  "invoice_amount": 6.39,
  "transactions": [{ "amount": 6.39, "transaction_type": "ReturnToSenderFee" }],
  "fulfillment_center": { "id": 178, "name": "Elwood (IL)" },
  "tracking_number": "...",
  "inventory": [{ "id", "name", "quantity", "action_taken": [...] }]
}

BigQuery Schema

Dataset

• Project: eli-health-prod
• Dataset: eli_health_shipbob
• Location: northamerica-northeast1 (Montreal)

Table Schema (consistent across all tables)

All tables use the raw JSON blob pattern (same as Loopwork):

Column	Type	Description
id	INT64	ShipBob entity ID (primary key)
data	JSON	Full API response as JSON blob
_synced_at	TIMESTAMP	When this row was last synced

Tables

1. orders — All ShipBob orders with nested shipments, products, recipient info
2. products — Product catalog with SKUs, pricing, inventory item mappings
3. inventory — Inventory levels per item per fulfillment center
4. returns — Return orders with transactions, actions taken
5. locations — ShipBob warehouse/fulfillment center reference data
6. channels — Sales channels (Shopify, Klaviyo, Zapier, PAT)

Example Queries

-- Get all fulfilled orders with tracking
SELECT
  id,
  JSON_VALUE(data, '$.order_number') AS order_number,
  JSON_VALUE(data, '$.status') AS status,
  JSON_VALUE(data, '$.recipient.name') AS customer_name,
  JSON_VALUE(data, '$.recipient.email') AS customer_email,
  JSON_VALUE(data, '$.shipments[0].tracking.tracking_number') AS tracking_number,
  TIMESTAMP(JSON_VALUE(data, '$.purchase_date')) AS purchase_date
FROM `eli-health-prod.eli_health_shipbob.orders`
WHERE JSON_VALUE(data, '$.status') = 'Fulfilled'
ORDER BY TIMESTAMP(JSON_VALUE(data, '$.purchase_date')) DESC;

-- Current inventory levels
SELECT
  id,
  JSON_VALUE(data, '$.name') AS item_name,
  CAST(JSON_VALUE(data, '$.total_sellable_quantity') AS INT64) AS sellable,
  CAST(JSON_VALUE(data, '$.total_onhand_quantity') AS INT64) AS onhand,
  CAST(JSON_VALUE(data, '$.total_committed_quantity') AS INT64) AS committed,
  CAST(JSON_VALUE(data, '$.total_backordered_quantity') AS INT64) AS backordered,
  JSON_VALUE(data, '$.is_active') AS is_active
FROM `eli-health-prod.eli_health_shipbob.inventory`
ORDER BY CAST(JSON_VALUE(data, '$.total_sellable_quantity') AS INT64) DESC;

-- Join ShipBob orders with Shopify orders via reference_id
SELECT
  sb.id AS shipbob_order_id,
  JSON_VALUE(sb.data, '$.order_number') AS shipbob_order_number,
  JSON_VALUE(sb.data, '$.status') AS fulfillment_status,
  JSON_VALUE(sb.data, '$.recipient.name') AS customer_name,
  JSON_VALUE(sb.data, '$.shipments[0].tracking.tracking_number') AS tracking
FROM `eli-health-prod.eli_health_shipbob.orders` sb
WHERE JSON_VALUE(sb.data, '$.channel.name') = 'elihealth'
ORDER BY sb.id DESC
LIMIT 20;

Sync Strategies

Orders: Rolling Window (7-day MERGE)

Orders change state over time (Processing → Fulfilled → Delivered), so we need incremental updates:

1. First run (empty table): Fetch ALL orders, paginate through all pages, INSERT
2. Subsequent runs:
   • Query orders updated in last 7 days: ?LastUpdateStartDate={7 days ago}
   • Create temp table, stream updated orders into it
   • MERGE temp table into orders table on id
   • Delete temp table

Small Tables: Full Refresh (TRUNCATE + INSERT)

Products, inventory, returns, locations, and channels are all small (fewer than 50 rows):

1. TRUNCATE table
2. Fetch all data from API (usually 1 page)
3. SQL INSERT with JSON blob

Implementation Details

Files

File	Action	Description
eli-kpi/src/shipbob-sync.ts	NEW	Main sync module (~400 lines)
eli-kpi/src/shipbob-routes.ts	NEW	Express routes (~50 lines)
eli-kpi/src/server.ts	MODIFY	Register ShipBob routes
eli-devops/tf/modules/regional/kpi-compute/main.tf	MODIFY	Cloud Scheduler job
eli-kpi/.env	MODIFY	Add SHIPBOB_API_TOKEN

Module Structure (shipbob-sync.ts)

ShipBobClient (class)
├── constructor(apiToken) — Axios instance with Bearer auth
├── rateLimiter — 400ms between requests
├── fetchAllPages(endpoint, params?) — Paginate until empty
└── fetchOrders(params?) — Orders with date filter support

syncShipBobData(apiToken?) → SyncResult
├── ensureDatasetExists()
├── ensureTableExists(tableName)
├── syncOrders() — Rolling window / full
├── syncFullRefreshTable('products', '/product')
├── syncFullRefreshTable('inventory', '/inventory')
├── syncFullRefreshTable('returns', '/return')
├── syncFullRefreshTable('locations', '/location')
├── syncFullRefreshTable('channels', '/channel')
└── return { success, message, stats, duration, errors }

getShipBobSyncStatus() → { lastSync, rowCounts }

Environment Variables

Variable	Value	Where
SHIPBOB_API_TOKEN	19FBF...130-1	.env (local), Secret Manager (prod)

Cloud Scheduler

Setting	Value
Name	shipbob-daily-sync
Schedule	0 3 * * * (3:00 AM UTC daily)
Timezone	America/Toronto
Endpoint	POST /sync/shipbob
Timeout	600s (10 minutes)
Retries	3 (exponential backoff: 30s → 300s)
Auth	OIDC token (KPI service account)

Sync Schedule (all times UTC)

Time	Sync Job
3:00 AM	ShipBob (new)
4:00 AM	Loopwork (subscriptions)
5:00 AM	Videnglobe Ads (Facebook/Google)

Verification

After Deployment

1. Trigger sync: POST /sync/shipbob
2. Check status: GET /sync/shipbob/status
3. Verify BigQuery dataset eli_health_shipbob has 6 tables
4. Verify row counts: orders ~3,500, products ~36, inventory ~21
5. Run sample queries from the Example Queries section above

BigQuery Verification Query

-- Check all ShipBob tables
SELECT
  table_id,
  row_count,
  TIMESTAMP_MILLIS(last_modified_time) as last_modified
FROM `eli-health-prod.eli_health_shipbob.__TABLES__`
ORDER BY table_id;