sync

Walk a collection page by page into your own database, resume where a run stopped, and ask only for records newer than the last finished run, with checkpoints in a store you supply.

sync({ checkpoints }) from outbound-sdk/sync adds outbound.sync.run(collection, { onPage }), for scheduled runs that keep your copy of a collection in step. It hands you each page and keeps one checkpoint per collection in a store you supply. The checkpoint holds a cursor, so a run that dies restarts on the page it was reading, and the start time of the last finished run, so the next run asks only for newer records. The SDK stores nothing itself and writes nothing to your database: onPage is where you do that.

Sample

The store below prints every checkpoint it is given. The run uses two leads per page on the shared seed.

import type { Checkpoint, CheckpointStore } from "outbound-sdk/sync";

// A checkpoint store is three async functions over a key and a small JSON
// value. This one keeps them in a Map and prints each write. In production
// the same three functions read and write a row in your database.
const saved = new Map<string, Checkpoint>();

export const checkpoints: CheckpointStore = {
  delete: (key) => {
    saved.delete(key);
    console.log("  checkpoint deleted");

    return Promise.resolve();
  },
  get: (key) => Promise.resolve(saved.get(key)),
  set: (key, checkpoint) => {
    saved.set(key, checkpoint);
    console.log("  checkpoint saved:", JSON.stringify(checkpoint));

    return Promise.resolve();
  },
};

import { createOutbound } from "outbound-sdk";
import type { Lead } from "outbound-sdk";
import { memory } from "outbound-sdk/memory";
import { sync } from "outbound-sdk/sync";

import { seed } from "../plugins-index/seed.ts";
import { checkpoints } from "./store.ts";

const outbound = createOutbound({
  // Two leads per page, so the four leads are two pages.
  adapter: memory({ data: seed, pageSize: 2, workspaceId: "agency" }),
  plugins: [sync({ checkpoints })],
});

// Your side of the sync: write the page to your database here.
const onPage = (leads: readonly Lead[]): void => {
  console.log("  page:", leads.map((lead) => lead.sourceId).join(", "));
};

console.log("run 1, stopped after one page:");

console.log(await outbound.sync.run<Lead>("leads", { maxPages: 1, onPage }));

console.log("run 2, picks up on the second page:");

console.log(await outbound.sync.run<Lead>("leads", { onPage }));

console.log(
  "run 3, asks only for leads created since run 1 began, less 5 minutes:"
);

console.log(await outbound.sync.run<Lead>("leads", { onPage }));

console.log("after reset, the next run starts over:");

await outbound.sync.reset("leads");

const again = await outbound.sync.run<Lead>("leads", { onPage });

console.log(again.records);

run 1, stopped after one page:
  page: lead_jane, lead_sam
  checkpoint saved: {"cursor":"eyJhIjoibGVhZHMubGlzdCIsIm4iOjIsInAiOiJtZW1vcnkiLCJ2IjoxLCJ3IjoiYWdlbmN5In0","startedAt":"2026-11-14T12:00:00.000Z"}
{
  collection: 'leads',
  complete: false,
  pages: 1,
  records: 2,
  resumed: false,
  since: undefined
}
run 2, picks up on the second page:
  page: lead_priya, lead_omar
  checkpoint saved: {"startedAt":"2026-11-14T12:00:00.000Z"}
  checkpoint saved: {"since":"2026-11-14T12:00:00.000Z"}
{
  collection: 'leads',
  complete: true,
  pages: 1,
  records: 2,
  resumed: true,
  since: undefined
}
run 3, asks only for leads created since run 1 began, less 5 minutes:
  checkpoint saved: {"since":"2026-11-14T12:00:00.000Z"}
{
  collection: 'leads',
  complete: true,
  pages: 0,
  records: 0,
  resumed: false,
  since: '2026-11-14T11:55:00.000Z'
}
after reset, the next run starts over:
  checkpoint deleted
  page: lead_jane, lead_sam
  checkpoint saved: {"cursor":"eyJhIjoibGVhZHMubGlzdCIsIm4iOjIsInAiOiJtZW1vcnkiLCJ2IjoxLCJ3IjoiYWdlbmN5In0","startedAt":"2026-11-14T12:00:00.000Z"}
  page: lead_priya, lead_omar
  checkpoint saved: {"startedAt":"2026-11-14T12:00:00.000Z"}
  checkpoint saved: {"since":"2026-11-14T12:00:00.000Z"}
4

Run 1 stopped at maxPages and left a cursor. Run 2 resumed from it and finished, which replaced the checkpoint with since: the time run 1 began. Run 3 asked for leads created since then, less the five-minute overlap, and found none. After reset, the next run read all four again.

Options

sync(options):

Option	Type	Default	Meaning
`checkpoints`	`CheckpointStore`	required	Where checkpoints are kept.
`overlapMinutes`	`number`	`5`	How far each run reaches back before the start of the last finished one, because records can land while a listing is being read.
`now`	`() => number`	`Date.now`	The clock, for tests.

outbound.sync.run(collection, options):

Option	Type	Default	Meaning
`collection`	`"campaigns" \| "leads" \| "messages" \| "replies" \| "senders"`	required	What to walk.
`onPage`	`(items) => void \| Promise<void>`	required	Receives each page. The checkpoint moves only after it resolves.
`campaignId`	`string`	none	Sync one campaign’s records. Not used for `campaigns`. It has its own checkpoint.
`maxPages`	`number`	none	Stop after this many pages. The next run resumes from the checkpoint.
`limit`	`number`	the adapter’s	Records per page, as a hint the adapter may clamp.

run is generic in the record type. Name it at the call site, as the sample does with run<Lead>, so onPage receives typed records. The name is not checked against collection.

What it adds

Member	Returns
`outbound.sync.run(collection, options)`	A `SyncResult`.
`outbound.sync.reset(collection, campaignId?)`	Deletes that checkpoint, so the next run starts over.

A SyncResult has:

Field	Meaning
`collection`	The collection that was walked.
`records`, `pages`	What this run delivered to `onPage`.
`resumed`	`true` when this run continued one that had not finished.
`since`	The lower bound this run asked the vendor for. `undefined` on a first run, and for a collection that cannot be filtered by date.
`complete`	`false` when the run stopped at `maxPages` with more to read.

How a run moves

It reads the checkpoint. A cursor means an earlier run did not finish, and this one starts from that page.
For leads, messages and replies, when the instance can filter that collection by date (leads.since, messages.since, replies.since is not false), it asks for records since the checkpoint’s since minus the overlap. campaigns and senders are always read in full.
Each time a page has been handed out in full, it awaits onPage, then saves the cursor for the next page.
When the listing ends, it replaces the checkpoint with { since }, set to when this run started. A run that resumed keeps the start time of the run it continued.

Three consequences:

A page can be delivered twice. A crash between onPage and the checkpoint write, and the overlap, both replay records. Write with an upsert keyed on ref.
An incremental run is not a change feed. It asks for records since a time. Where the SDK applies that filter itself, it compares a lead’s createdAt and a message’s sentAt, so a lead whose status changed later is not picked up. A deleted record is never reported. Call reset for a full pass, and use reconcile to find deletions.
The first run reads everything.

The store

A checkpoint store is three async functions. A checkpoint is a small JSON object with up to three strings: cursor, since and startedAt.

Function	Called
`get(key)`	Once at the start of each run. Return `undefined` when there is none.
`set(key, checkpoint)`	After every page, and once when the run finishes.
`delete(key)`	By `sync.reset`.

The key is outbound-sync: followed by a hash of the instance’s provider, workspace label, the collection and the campaignId. Two instances with the same provider and workspace label share checkpoints, which is what lets a new process resume.

memoryCheckpointStore() keeps checkpoints in this process’s memory. It is meant for tests and one-off scripts: a checkpoint that does not outlive the process cannot resume anything.

What it changes

It changes no capability, no record and nothing in meta. It has no layer of its own, so its position in plugins does not matter.

With other plugins

run reads through the instance’s public API, so every layer applies to the pages it delivers.

With redact, onPage receives masked records.
With cache, a page can be served from the store. Leave the list operations out of the cache (only) on an instance you sync from, or give the sync its own instance.
A budget or maxRequests cap that stops a run leaves the checkpoint at the last page that was handled. The next run resumes there.

When it fails

import { createOutbound, OutboundError } from "outbound-sdk";
import type { Lead, Message } from "outbound-sdk";
import { memory } from "outbound-sdk/memory";
import { memoryCheckpointStore, sync } from "outbound-sdk/sync";

import { seed } from "../plugins-index/seed.ts";

const outbound = createOutbound({
  adapter: memory({ data: seed, pageSize: 2, workspaceId: "agency" }),
  plugins: [sync({ checkpoints: memoryCheckpointStore() })],
});

// 1. Your database is down while the second page is being written.
let databaseUp = false;
const delivered: string[] = [];

const onPage = (leads: readonly Lead[]): void => {
  if (!databaseUp && delivered.length > 0) {
    throw new Error("database is down");
  }

  delivered.push(leads.map((lead) => lead.sourceId).join(", "));
};

try {
  await outbound.sync.run<Lead>("leads", { onPage });
} catch (error) {
  console.log("run 1 failed:", error instanceof Error ? error.message : error);
}

// The checkpoint is still before the page that failed, so it comes again.
databaseUp = true;

const retry = await outbound.sync.run<Lead>("leads", { onPage });

console.log("run 2 resumed:", retry.resumed, "pages:", retry.pages);

console.log("delivered:", delivered);

// 2. An adapter that cannot filter messages by date: every run reads them all.
const noDates = createOutbound({
  adapter: memory({
    capabilities: { "messages.since": false },
    data: seed,
    workspaceId: "agency",
  }),
  plugins: [sync({ checkpoints: memoryCheckpointStore() })],
});

const ignore = (_page: readonly Message[]): void => {
  // A real sync writes the page somewhere.
};

await noDates.sync.run<Message>("messages", { onPage: ignore });

const second = await noDates.sync.run<Message>("messages", { onPage: ignore });

console.log("no date filter:", second.since, second.records);

// 3. An adapter that cannot list replies refuses the sync.
const noReplies = createOutbound({
  adapter: memory({ capabilities: { "replies.list": false }, data: seed }),
  plugins: [sync({ checkpoints: memoryCheckpointStore() })],
});

try {
  await noReplies.sync.run("replies", { onPage: ignore });
} catch (error) {
  if (OutboundError.is(error)) {
    console.log("no replies:", error.code, "-", error.message);
  }
}

run 1 failed: database is down
run 2 resumed: true pages: 1
delivered: [ 'lead_jane, lead_sam', 'lead_priya, lead_omar' ]
no date filter: undefined 3
no replies: Unsupported - Listing replies is not available: the memory adapter does not support it (replies.list)

What happened	What you get	What to do
`onPage` throws or rejects	`run` rejects with that error. The checkpoint is still before the page.	Fix the cause and run again. The page is delivered again.
A vendor read fails part way	`run` rejects with the `OutboundError`. Pages already handled are checkpointed.	Run again. It resumes.
The adapter cannot filter the collection by date	Every run reads the whole collection. `since` in the result is `undefined`.	Expect the cost of a full walk each time, and upsert.
The adapter cannot list the collection	`Unsupported`, before any request	Sync a collection the adapter declares.
The name is not one of the five collections	`Invalid`, before any read and before the checkpoint is touched. `reset` refuses it too.	Fix the name. Code without types can pass any string.
The store rejects	`run` rejects with the store’s error	Run again once the store is back.