Chapter 4: Managing state of your Integration

A schema that covers all four (SQLite or Postgres syntax):

-- Every table is scoped to one integration: (org_key, integration_id)

-- Where each poll stream left off
CREATE TABLE cursors (
  org_key        TEXT NOT NULL,
  integration_id TEXT NOT NULL,
  stream         TEXT NOT NULL,          -- e.g. 'shipping_requests.ready', 'products'
  updated_since  TEXT NOT NULL,          -- RFC 3339 timestamp
  last_run_at    TEXT,
  PRIMARY KEY (org_key, integration_id, stream)
);

-- Pipe17 record <-> your record
CREATE TABLE id_map (
  org_key        TEXT NOT NULL,
  integration_id TEXT NOT NULL,
  entity         TEXT NOT NULL,          -- 'shipments', 'locations', 'products', 'fulfillments'
  pipe17_id      TEXT,
  ext_id         TEXT,
  last_status    TEXT,                   -- last status you saw or set
  last_version   TEXT,                   -- Pipe17 version or updatedAt you last processed
  data           TEXT,                   -- optional JSON snapshot for diffing
  updated_at     TEXT NOT NULL,
  PRIMARY KEY (org_key, integration_id, entity, pipe17_id),
  UNIQUE (org_key, integration_id, entity, ext_id)
);

-- Idempotency: one row per change you have fully handled
CREATE TABLE processed (
  org_key        TEXT NOT NULL,
  integration_id TEXT NOT NULL,
  event_key      TEXT NOT NULL,
  processed_at   TEXT NOT NULL,
  PRIMARY KEY (org_key, integration_id, event_key)
);

-- Work that failed and must be retried
CREATE TABLE retry_queue (
  id             INTEGER PRIMARY KEY,
  org_key        TEXT NOT NULL,
  integration_id TEXT NOT NULL,
  job_type       TEXT NOT NULL,          -- 'pushFulfillment', 'sendShippingRequest', ...
  payload        TEXT NOT NULL,          -- JSON
  attempts       INTEGER NOT NULL DEFAULT 0,
  next_try_at    TEXT NOT NULL,
  last_error     TEXT
);

Rules that keep state correct

• Advance a cursor only after the whole window is handled. Items that failed go to the retry queue, so the cursor can still move forward.
• Write the mapping before you call the other side, then complete it. For example, insert id_map with pipe17_id and no ext_id, create the order in your system, then save ext_id. If the process crashes in between, the half-filled row tells you to check your system before creating again.
• Make every write idempotent. Build an event key from the record ID plus its version (or updatedAt), skip keys already in processed, and insert the key only after the work succeeds.
• Prefer lookups over assumptions. Before creating a Fulfillment, look it up by extFulfillmentId; before creating a Location, look it up in id_map. Pipe17 rejects duplicate keys with HTTP 409, which you can treat as "already done".
• Keep secrets out of the state store. Store API keys in your secrets manager, not alongside cursors.

A thin data-access module keeps the rest of the code simple. The examples in later chapters assume these functions exist: getCursor(stream), setCursor(stream, ts), getMapping(entity, pipe17Id), findByExtId(entity, extId), findByPipe17Id(entity, pipe17Id), saveMapping(entity, row), wasProcessed(key), markProcessed(key) and enqueueRetry(jobType, payload, error).

Troubleshooting

• Locations don't appear in the app
  Confirm that the locations.in setting is turned on and that your job runs without errors. Check the job's task status for failures.
• Duplicate locations appear after each run
  Your code is creating locations instead of updating them. Store the location ID the app returns, or check the List locations operation before you create a location.
• Orders aren't routed to your locations
  Confirm that fulfillmentIntegrationId on each location is set to your integration ID. In the app, the location's 3PL Integration must show your custom integration.

Need Help?

If you need additional assistance:

• Use Ask Pippen, our AI agent, located at the top of the app page.
• Submit a support request with as much relevant detail as possible. Learn how to submit a request.

We're here to help you succeed with your operations.