Chapter 5: Polling Pipe17

The polling pattern

1. Read the stream's cursor. On the very first run, start from a sensible point such as 24 hours ago.
2. Set updatedSince to the cursor minus a small overlap (2 minutes) to absorb clock skew and records written while you were paging.
3. Set updatedUntil to the time the run started, so the window does not move while you page.
4. Page with count and skip, sorted by updatedAt, until a page returns fewer than count records.
5. For each record, skip it if its event key is already processed; otherwise handle it.
6. When every page is done, set the cursor to the run start time.

// poll.js – generic incremental poller
import { p17 } from './p17.js';
import * as state from './state.js';

const PAGE = 100;
const OVERLAP_MS = 2 * 60 * 1000;

export async function pollStream({ stream, path, listKey, idField, query = {}, handle }) {
  const runStartedAt = new Date().toISOString();
  const cursor = state.getCursor(stream) ?? new Date(Date.now() - 24 * 3600e3).toISOString();
  const updatedSince = new Date(Date.parse(cursor) - OVERLAP_MS).toISOString();

  for (let skip = 0; ; skip += PAGE) {
    const page = await p17('GET', path, {
      query: { ...query, updatedSince, updatedUntil: runStartedAt, order: 'updatedAt', count: PAGE, skip, keys: '*' },
    });
    const items = page[listKey] ?? [];

    for (const item of items) {
      const eventKey = `${stream}:${item[idField]}:${item.version ?? item.updatedAt}`;
      if (state.wasProcessed(eventKey)) continue;
      try {
        await handle(item);
        state.markProcessed(eventKey);
      } catch (err) {
        state.enqueueRetry(stream, { id: item[idField] }, err);   // retried later, cursor still moves
      }
    }
    if (items.length < PAGE) break;
  }

  state.setCursor(stream, runStartedAt);
}

The same request as curl:

curl -s -G "$PIPE17_BASE_URL/shipping_requests" \
  -H "X-Pipe17-Key: $PIPE17_INTEGRATION_KEY" \
  --data-urlencode "status=readyForFulfillment" \
  --data-urlencode "fulfillmentIntegrationId=<your integrationId>" \
  --data-urlencode "updatedSince=2026-10-12T06:53:00Z" \
  --data-urlencode "updatedUntil=2026-10-12T07:00:00Z" \
  --data-urlencode "order=updatedAt" \
  --data-urlencode "count=100" \
  --data-urlencode "skip=0" \
  --data-urlencode "keys=*"

Streams a 3PL connector polls

fulfillmentIntegrationId returns only the Shipping Requests routed to your integration, which is simpler and safer than filtering by location. Set updatedUntil about a minute in the past so records still being written are picked up on the next run.

Your own system needs polling too, for shipments and stock levels it does not push to you. Use the same cursor table with stream names such as wms.shipments and wms.inventory.

Polling tips

• Run one copy of each job at a time. Use a lock row or your scheduler's concurrency setting so two runs never share a window.
• Ask for what you need. Default list responses return only a few fields. Pass keys=* or a comma-separated field list to get line items and addresses.
• Use updatedSince, not since. since filters by creation time and misses changes to existing records.
• Expect to see a record more than once. The overlap and webhooks (Chapter 6) both cause repeats; the event key makes them harmless.

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.