Sync routes

Per-stream poll handlers that fetch from the provider and emit canonical records.

For every poll-mode stream you declare, write a sync route. The platform invokes it on schedule, hands you the saved page/cursor, and expects you to fetch from the provider and emit canonical records via ingest.canonical.

File location

Sync routes live at src/api/sync/<stream-key>.ts — auto-registered as the POST handler for the matching streams[] entry in poll or backfill mode. This is an enforced build/deploy contract: every poll or backfill stream must have a matching POST sync route. They follow the same conventions as plain API routes.

Real example

A real sync route — Stripe customers:

import { api, ingest } from "@backfill-io/sdk";
import { stripePollMetadata } from "../../lib/metadata";
import {
  commitStripeListPage,
  missingStripeApiKeyResponse,
  stripeApiKey,
  stripeDateFilterParams,
  stripeGet,
  stripeListPage,
  stripeListResponse,
  stripeListUrl,
  stripeSyncResponse,
} from "../../lib/stripe";

export const config = { auth: "api_key" };

export const POST = api(async (request) => {
  const stream = "customers";
  const apiKey = stripeApiKey();
  if (!apiKey) return missingStripeApiKeyResponse();

  const page = stripeListPage(request, stream);
  const response = stripeGet(
    stripeListUrl("/customers", page, stripeDateFilterParams(request)),
    apiKey,
  );
  if (!response.ok) {
    return api.providerError("Stripe customers sync failed", response);
  }

  const list = stripeListResponse(response);
  const payloads = list.records.map((customer) => customerCanonical(customer));
  const batch = ingest.batch("customer", payloads, {
    stream,
    continueOnError: true,
    sourceId: (payload) => payload.customer_id,
    idempotencyKey: (payload) => `stripe:customer:${payload.customer_id}`,
  });

  const checkpoint = commitStripeListPage(stream, list.records, list.hasMore);

  return stripeSyncResponse({
    stream,
    imported: batch.imported,
    failed: batch.failed,
    hasMore: list.hasMore,
    checkpoint,
    results: batch.results,
    failedRecords: batch.failedRecords,
  });
});

The contract

A sync route is just an API route with POST. The platform invokes it as part of the schedule for the matching stream.

Concern	Where it lives
Scheduling	The `streams[]` entry in the manifest (`schedule: "0 /6 * *"`), unless the stream opts into `scheduleConfigurable: true` and the admin chooses a per-connection override.
Auth into the provider	`Secrets.get("api_key")` (or whatever your `settingsSchema` declared).
Cursor / page	Read from `request` (e.g., `searchParams`) and persisted between runs by the platform / lib helpers.
Outbound HTTP	`Http` (with the provider’s host on `permissions.http`).
Emit	`ingest.batch(canonicalType, payloads, { stream, continueOnError: true, sourceId, idempotencyKey })` for pages, or `ingest.canonical(...)` for one record.
Telemetry	`Log.info("...", { ... })`.
Response	A status object the platform reads to know how the run went.

`ingest.canonical`

The connector-specific runtime helper. Signature observed from usage:

ingest.canonical(
  canonicalType: string,                // canonical entity name (lowercase or platform-mapped)
  payload: Record<string, any>,         // shaped to canonical schema
  opts: { stream: string; idempotencyKey: string }
): IngestResult;

Argument	Notes
`canonicalType`	The canonical entity to write to (e.g., `"customer"`, `"invoice"`). The platform routes this to the right concrete entity for the connection.
`payload`	Shape it as the canonical schema expects, not the provider’s raw shape. Most connectors maintain `lib/transforms/*.ts` to normalize.
`stream`	The stream key from `streams[]`. Used for telemetry + checkpoint accounting.
`idempotencyKey`	A stable string derived from the provider’s record. Re-emissions with the same key dedupe.

`ingest.batch`

For sync pages, prefer ingest.batch(..., { continueOnError: true }) so one bad record does not abort the whole page:

const batch = ingest.batch("customer", payloads, {
  stream: "customers",
  continueOnError: true,
  sourceId: (payload) => payload.customer_id,
  idempotencyKey: (payload) => `stripe:customer:${payload.customer_id}`,
});

The result includes:

Field	Notes
`imported`	Successful records.
`failed`	Failed records.
`results`	Successful canonical ingest results.
`failedRecords`	Structured failures with `documentType`, `stream`, `sourceId`, `idempotencyKey`, `reason`, `message`, `retryable`, and `index`.

Pagination & checkpoints

Use sync.listPage(...) to read the saved cursor and request-level pagination overrides, then sync.commitListPage(...) after the page has been ingested.

const page = sync.listPage(request, "transactions", {
  defaultLimit: 1000,
  maxLimit: 1000,
  requestCursorKeys: ["starting_after", "startingAfter"],
  checkpointCursorKeys: ["startingAfter", "cursorValue"],
});

const response = Http.get("https://api.example.com/v1/transactions", {
  auth: { bearer: Secrets.get("api_key")! },
  query: {
    limit: page.limit,
    starting_after: page.cursor,
  },
});

const checkpoint = sync.commitListPage("transactions", response.body.data, {
  cursorField: "id",
  hasMore: response.body.has_more === true,
});

Checkpoint partitions

Use a checkpoint partition when the provider cursor depends on user-selected filters such as accounts, cards, date windows, or regions. Partitions let one stream keep independent cursors for different import scopes.

const partition = {
  accountIds: [...selectedAccountIds].sort(),
  cardIds: [...selectedCardIds].sort(),
  postedStart: importStartDate || null,
};

const page = sync.listPage(request, "bank_transactions", {
  defaultLimit: 1000,
  maxLimit: 1000,
  requestCursorKeys: ["start_after", "startAfter"],
  checkpointCursorKeys: ["startAfter", "cursorValue", "pageToken"],
  partition,
});

const response = Http.get("https://api.example.com/api/v1/transactions", {
  auth: { bearer: Secrets.get("api_token")! },
  query: {
    limit: page.limit,
    order: "asc",
    start_after: page.cursor,
    postedStart: importStartDate,
    cardId: selectedCardIds,
  },
});

const checkpoint = sync.commitListPage("bank_transactions", response.body.transactions, {
  cursorField: "id",
  cursorValue: response.body.page?.nextPage,
  hasMore: response.body.page?.hasMore === true,
  partition,
});

Keep partition objects small and deterministic. Sort arrays before passing them. Do not use partitions for ordinary provider pagination when the cursor does not depend on connector configuration.

Errors

If the provider call fails, return early with a structured response (e.g., api.providerError(...)). Don’t throw — ingest.canonical is not aware of mid-pagination throws and you may end up with partial state.

If a single record fails to ingest, log the failure and continue the page; the platform records ingest results per record so partial success is observable.

Testing sync routes

Use @backfill-io/sdk/testing to run sync routes locally with mocked provider HTTP, settings, secrets, checkpoints, and ingest calls:

import assert from "node:assert/strict";
import test from "node:test";
import { connectorTestRuntime } from "@backfill-io/sdk/testing";

test("customers sync emits canonical customers", async () => {
  const runtime = connectorTestRuntime({
    secrets: { api_key: "sk_test_123" },
  });

  try {
    runtime.install();
    const { POST } = await import("../src/api/sync/customers");

    runtime.http.getOnce({
      status: 200,
      body: { data: [{ id: "cus_123", name: "Ada Lovelace" }] },
    });

    const response = await runtime.call(POST);

    assert.equal(response.status, 200);
    assert.equal(runtime.ingest.calls[0].documentType, "customer");
    assert.equal(runtime.ingest.calls[0].sourceId, "cus_123");
  } finally {
    runtime.restore();
  }
});

See Testing for the full route-test harness, including partial ingest failures and pagination assertions.

← Sync API

Settings schema →