Keep a copy of campaigns, leads and replies in your database

A scheduled job that upserts campaigns, leads and replies into your own tables, asks only for what is new where the vendor can filter by date, and restarts on the page it was writing after a crash.

Dashboards and joins want the data in your own database, not behind a vendor’s rate limit. The sync plugin walks a collection page by page, hands each page to a function you write, and keeps one checkpoint per collection in a store you supply: 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 plugin writes nothing itself. This guide supplies the three parts it leaves to you: the tables, the upserts and the checkpoint store. It also shows what a sync does not do: it does not copy stats, and it does not notice a record that was deleted at the vendor.

Before you start

• Node 22 or later. The sample database is node:sqlite, which is built into Node; the samples are run on Node 24, where it needs no flag.
• The code below runs on the in-memory adapter, so it needs no vendor key. Each sample on this page is a file in the repository that the test suite runs; the output shown is what it printed.
• For real data: an Instantly, Smartlead or HeyReach API key. A read-only key is enough. No method of the SDK writes to a vendor.

What you get

Two runs against the same database, an hour apart. Between them one lead and one reply were added at the vendor.

First run
  campaigns: 2 record(s) in 1 page(s), lower bound none
  leads: 5 record(s) in 1 page(s), lower bound none
  replies: 2 record(s) in 1 page(s), lower bound none
  rows: 2 campaigns, 5 leads, 2 replies

Second run
  campaigns: 2 record(s) in 1 page(s), lower bound none
  leads: 1 record(s) in 1 page(s), lower bound 2026-11-14T11:55:00.000Z
  replies: 1 record(s) in 1 page(s), lower bound 2026-11-14T11:55:00.000Z
  rows: 2 campaigns, 6 leads, 3 replies

The first run reads everything. The second asks for leads and replies from five minutes before the first run started, and gets one of each. Campaigns have no date filter, so they are listed in full on every run.

The tables and the upserts

Every record carries a ref (provider:workspace:id) that is unique across vendors and clients. Use it as the primary key.

// The copy: three tables keyed on `ref`, and the sync checkpoints beside
// them. `node:sqlite` is built into Node; an in-memory database keeps the
// sample self-contained.
import { DatabaseSync } from "node:sqlite";
import type { SQLOutputValue } from "node:sqlite";

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

export const openDatabase = (): DatabaseSync => {
  const db = new DatabaseSync(":memory:");

  db.exec(`
    CREATE TABLE campaigns (ref TEXT PRIMARY KEY, name TEXT, status TEXT);
    CREATE TABLE leads (
      ref TEXT PRIMARY KEY, campaign_ref TEXT, email TEXT, status TEXT
    );
    CREATE TABLE replies (
      ref TEXT PRIMARY KEY, lead_ref TEXT, category TEXT, sent_at TEXT
    );
    CREATE TABLE sync_checkpoints (
      key TEXT PRIMARY KEY, cursor TEXT, since TEXT, started_at TEXT
    );
  `);

  return db;
};

// One page is one transaction: it is written whole or not at all.
const inTransaction = (db: DatabaseSync, write: () => void): void => {
  db.exec("BEGIN");

  try {
    write();
    db.exec("COMMIT");
  } catch (error) {
    db.exec("ROLLBACK");
    throw error;
  }
};

// Every write is an upsert on `ref`, so a page delivered twice changes nothing.
export const upsertCampaigns = (
  db: DatabaseSync,
  campaigns: readonly Campaign[]
): void => {
  const upsert = db.prepare(`
    INSERT INTO campaigns (ref, name, status) VALUES (?, ?, ?)
    ON CONFLICT (ref) DO UPDATE SET
      name = excluded.name, status = excluded.status
  `);

  inTransaction(db, () => {
    for (const campaign of campaigns) {
      upsert.run(campaign.ref, campaign.name, campaign.status);
    }
  });
};

export const upsertLeads = (db: DatabaseSync, leads: readonly Lead[]): void => {
  const upsert = db.prepare(`
    INSERT INTO leads (ref, campaign_ref, email, status) VALUES (?, ?, ?, ?)
    ON CONFLICT (ref) DO UPDATE SET
      campaign_ref = excluded.campaign_ref,
      email = excluded.email,
      status = excluded.status
  `);

  inTransaction(db, () => {
    for (const lead of leads) {
      upsert.run(lead.ref, lead.campaignRef, lead.email, lead.status);
    }
  });
};

export const upsertReplies = (
  db: DatabaseSync,
  replies: readonly Reply[]
): void => {
  const upsert = db.prepare(`
    INSERT INTO replies (ref, lead_ref, category, sent_at) VALUES (?, ?, ?, ?)
    ON CONFLICT (ref) DO UPDATE SET
      lead_ref = excluded.lead_ref,
      category = excluded.category,
      sent_at = excluded.sent_at
  `);

  inTransaction(db, () => {
    for (const reply of replies) {
      upsert.run(reply.ref, reply.leadRef, reply.category, reply.sentAt);
    }
  });
};

// A column is text or NULL. The store's fields are text or absent.
const text = (value: SQLOutputValue | undefined): string | undefined =>
  value === null || value === undefined ? undefined : String(value);

/** The plugin's checkpoints, kept in the database the records are written to. */
export const sqliteCheckpoints = (db: DatabaseSync): CheckpointStore => ({
  delete: (key) => {
    db.prepare("DELETE FROM sync_checkpoints WHERE key = ?").run(key);

    return Promise.resolve();
  },
  get: (key) => {
    const row = db
      .prepare(
        "SELECT cursor, since, started_at FROM sync_checkpoints WHERE key = ?"
      )
      .get(key);

    const checkpoint: Checkpoint | undefined =
      row === undefined
        ? undefined
        : {
            cursor: text(row.cursor),
            since: text(row.since),
            startedAt: text(row.started_at),
          };

    return Promise.resolve(checkpoint);
  },
  set: (key, checkpoint) => {
    db.prepare(
      `INSERT INTO sync_checkpoints (key, cursor, since, started_at)
       VALUES (?, ?, ?, ?)
       ON CONFLICT (key) DO UPDATE SET
         cursor = excluded.cursor,
         since = excluded.since,
         started_at = excluded.started_at`
    ).run(
      key,
      checkpoint.cursor ?? null,
      checkpoint.since ?? null,
      checkpoint.startedAt ?? null
    );

    return Promise.resolve();
  },
});

export const countRows = (db: DatabaseSync, table: string): number => {
  const row = db.prepare(`SELECT count(*) AS total FROM ${table}`).get();

  return Number(row?.total ?? 0);
};

Two things here are not optional:

• Every write is an upsert. The plugin moves its checkpoint only after your onPage function resolves. If the process dies after your write and before the checkpoint is saved, the next run delivers that page again. The overlap between runs delivers records twice as well. An INSERT without ON CONFLICT would fail or duplicate rows.
• The checkpoints are in the same database as the records. A checkpoint kept somewhere that can be lost separately (a file on a container’s disk, a cache) lets the two drift. A CheckpointStore is three functions: get, set and delete. The key is an opaque string that already includes the provider, the workspace, the collection and the campaign.

The same upsert in Postgres. This statement is not run by the test suite:

INSERT INTO leads (ref, campaign_ref, email, status)
VALUES ($1, $2, $3, $4)
ON CONFLICT (ref) DO UPDATE SET
  campaign_ref = EXCLUDED.campaign_ref,
  email = EXCLUDED.email,
  status = EXCLUDED.status;

The sync

outbound.sync.run(collection, options) syncs one collection: campaigns, leads, replies, messages or senders. The type argument names the record your onPage receives. It is not checked against the collection, so keep the two in step.

import type { DatabaseSync } from "node:sqlite";

import { createOutbound } from "outbound-sdk";
import type { Adapter, Campaign, Lead, Reply } from "outbound-sdk";
import { sync } from "outbound-sdk/sync";
import type { SyncResult } from "outbound-sdk/sync";

import {
  sqliteCheckpoints,
  upsertCampaigns,
  upsertLeads,
  upsertReplies,
} from "./database.ts";

/** An instance whose sync checkpoints live in `db`. `now` is the clock. */
export const syncing = (
  adapter: Adapter,
  db: DatabaseSync,
  now?: () => number
) =>
  createOutbound({
    adapter,
    plugins: [sync({ checkpoints: sqliteCheckpoints(db), now })],
  });

/** Copies campaigns, leads and replies into `db`. Run it on a schedule. */
export const syncAll = async (
  outbound: ReturnType<typeof syncing>,
  db: DatabaseSync
): Promise<SyncResult[]> => [
  await outbound.sync.run<Campaign>("campaigns", {
    onPage: (campaigns) => upsertCampaigns(db, campaigns),
  }),
  await outbound.sync.run<Lead>("leads", {
    onPage: (leads) => upsertLeads(db, leads),
  }),
  await outbound.sync.run<Reply>("replies", {
    onPage: (replies) => upsertReplies(db, replies),
  }),
];

export const describe = (result: SyncResult): string =>
  `${result.collection}: ${result.records} record(s) in ${result.pages} page(s), lower bound ${result.since ?? "none"}`;

Each run returns a SyncResult: how many records and pages it read, whether it continued an unfinished run (resumed), the lower bound it asked for (since) and whether it reached the end (complete).

How the lower bound works:

• A run that finishes saves the time it started. The next run asks for records from that time minus overlapMinutes (default 5), because a record can land while a listing is in progress.
• Only leads, replies and messages can be filtered by date. campaigns and senders are listed in full every run.
• For leads the date is when the lead was created. A lead whose status changed is not “newer”, so a dated sync does not pick the change up. Call outbound.sync.reset("leads") before a run to read every lead again, as often as you need statuses to be current.

Try it without a key

// One client, "acme": two campaigns, five leads and their messages. `later`
// is the same account an hour on: one new lead, one new reply, one lead gone.
import type {
  MemoryData,
  MemoryLead,
  MemoryMessage,
} from "outbound-sdk/memory";

const lead = (
  id: string,
  campaignId: string,
  email: string,
  createdAt: string
): MemoryLead => ({ campaignId, createdAt, email, id });

const sent = (id: string, leadId: string, sentAt: string): MemoryMessage => ({
  campaignId: "cmp_founders",
  direction: "out",
  id,
  leadId,
  sentAt,
});

const reply = (
  id: string,
  leadId: string,
  sentAt: string,
  category: "negative" | "positive"
): MemoryMessage => ({
  campaignId: "cmp_founders",
  category,
  direction: "in",
  id,
  leadId,
  sentAt,
});

const LEADS: readonly MemoryLead[] = [
  lead("lead_1", "cmp_founders", "ana@example.com", "2026-11-10T09:00:00Z"),
  lead("lead_2", "cmp_founders", "ben@example.com", "2026-11-10T09:00:00Z"),
  lead("lead_3", "cmp_founders", "caro@example.com", "2026-11-11T09:00:00Z"),
  lead("lead_4", "cmp_agencies", "dev@example.com", "2026-11-12T09:00:00Z"),
  lead("lead_5", "cmp_agencies", "eli@example.com", "2026-11-13T09:00:00Z"),
];

const MESSAGES: readonly MemoryMessage[] = [
  sent("msg_1", "lead_1", "2026-11-11T08:00:00Z"),
  sent("msg_2", "lead_2", "2026-11-11T08:05:00Z"),
  reply("msg_3", "lead_1", "2026-11-12T10:30:00Z", "positive"),
  reply("msg_4", "lead_2", "2026-11-13T16:45:00Z", "negative"),
];

const CAMPAIGNS = [
  { id: "cmp_founders", name: "SaaS founders" },
  { id: "cmp_agencies", name: "Agency owners" },
];

export const seed: MemoryData = {
  campaigns: CAMPAIGNS,
  leads: LEADS,
  messages: MESSAGES,
};

export const later: MemoryData = {
  campaigns: CAMPAIGNS,
  leads: [
    // lead_2 was deleted in the vendor's app.
    ...LEADS.filter((item) => item.id !== "lead_2"),
    lead("lead_6", "cmp_agencies", "fay@example.com", "2026-11-14T12:20:00Z"),
  ],
  messages: [
    ...MESSAGES,
    reply("msg_5", "lead_3", "2026-11-14T12:40:00Z", "positive"),
  ],
};

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

import { countRows, openDatabase } from "./database.ts";
import { later, seed } from "./seed.ts";
import { describe, syncAll, syncing } from "./sync-all.ts";

const db = openDatabase();
const HOUR = 3_600_000;

const report = (title: string, results: readonly SyncResult[]): void => {
  console.log(title);

  for (const result of results) {
    console.log(`  ${describe(result)}`);
  }

  console.log(
    `  rows: ${countRows(db, "campaigns")} campaigns, ${countRows(db, "leads")} leads, ${countRows(db, "replies")} replies\n`
  );
};

// 12:00. Nothing has been synced yet, so every record is read.
const first = syncing(memory({ data: seed, workspaceId: "acme" }), db);

report("First run", await syncAll(first, db));

// 13:00. The same account an hour later, and the same database.
const second = syncing(
  memory({ data: later, workspaceId: "acme" }),
  db,
  () => Date.now() + HOUR
);

report("Second run", await syncAll(second, db));

That is the script that printed the two runs above. The fixed clock of the samples is 12:00, so the second instance is given a clock one hour ahead.

A run that dies resumes on the page it was writing

Here onPage writes the second page and then throws, which is what a lost connection or a killed process looks like to the plugin.

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

import { countRows, openDatabase, upsertLeads } from "./database.ts";
import { seed } from "./seed.ts";
import { syncing } from "./sync-all.ts";

const db = openDatabase();

// Two leads a page, so five leads are three pages.
const outbound = syncing(
  memory({ data: seed, pageSize: 2, workspaceId: "acme" }),
  db
);

let pages = 0;

// The second page is written, and then the process "dies" before the plugin
// can move its checkpoint.
try {
  await outbound.sync.run<Lead>("leads", {
    onPage: (leads) => {
      upsertLeads(db, leads);
      pages += 1;

      if (pages === 2) {
        throw new Error("connection lost");
      }
    },
  });
} catch (error) {
  console.log(`first run failed: ${String(error)}`);
}

console.log(`leads written so far: ${countRows(db, "leads")}`);

// The next run starts on the page that failed, not on the first one.
const resumed = await outbound.sync.run<Lead>("leads", {
  onPage: (leads) => {
    console.log(`  page: ${leads.map((lead) => lead.sourceId).join(", ")}`);
    upsertLeads(db, leads);
  },
});

console.log(
  `second run: resumed ${resumed.resumed}, ${resumed.records} records in ${resumed.pages} pages`
);

console.log(`leads in the database: ${countRows(db, "leads")}`);

first run failed: Error: connection lost
leads written so far: 4
  page: lead_3, lead_4
  page: lead_5
second run: resumed true, 3 records in 2 pages
leads in the database: 5

The second run starts with lead_3, not lead_1. lead_3 and lead_4 were already written by the run that failed, and they are delivered again. The upsert makes that harmless: five leads, five rows.

Point it at your vendor

Swap the adapter. The tables, the upserts and the checkpoint store do not change.

import { createOutbound } from "outbound-sdk";
import { instantly } from "outbound-sdk/instantly";
import { sync } from "outbound-sdk/sync";

import { openDatabase, sqliteCheckpoints } from "./database.ts";

// Open your real database here. The checkpoints must outlive the process.
export const db = openDatabase();

// Reads INSTANTLY_API_KEY. Swap the import for smartlead() or heyreach().
export const outbound = createOutbound({
  adapter: instantly({ workspaceId: "acme" }),
  plugins: [sync({ checkpoints: sqliteCheckpoints(db) })],
});

Then call syncAll from whatever scheduler you already run (cron, a GitHub Actions schedule, a Vercel cron route). The SDK schedules nothing. To bound one invocation, pass maxPages: the run stops with complete: false and the next invocation continues from the checkpoint.

Do not start two runs of the same collection at once. They would read the same checkpoint and overwrite each other’s.

Stats are not a collection

There is no sync.run("stats"): a name that is not a collection is refused as Invalid. Stats are totals the vendor computes, not records with a ref, and a vendor can revise yesterday’s numbers. Read them again on each run with outbound.campaign(id).stats({ by: "day", from, to }) for the days you keep, and overwrite your rows for those days.

Deleted records are not detected

A sync only ever sees what the vendor lists. When a lead is deleted at the vendor it stops appearing, and nothing tells the sync. After the second run above, the database still holds the deleted lead.

The reconcile plugin’s deletions() lists the whole collection and compares it with the refs you hold:

import { createOutbound } from "outbound-sdk";
import { memory } from "outbound-sdk/memory";
import { reconcile } from "outbound-sdk/reconcile";

import { openDatabase } from "./database.ts";
import { later, seed } from "./seed.ts";
import { syncAll, syncing } from "./sync-all.ts";

const db = openDatabase();

await syncAll(syncing(memory({ data: seed, workspaceId: "acme" }), db), db);

// An hour later lead_2 is gone from the vendor. No sync run will say so.
const outbound = createOutbound({
  adapter: memory({ data: later, workspaceId: "acme" }),
  plugins: [reconcile()],
});

const known = db
  .prepare("SELECT ref FROM leads")
  .all()
  .map((row) => String(row.ref));

const report = await outbound.reconcile.deletions({
  collection: "leads",
  known,
});

console.log(`listed ${report.listed} leads`);

console.log("deleted at the vendor:", report.deleted);

console.log("not in the database yet:", report.added);

listed 5 leads
deleted at the vendor: [ 'memory:acme:lead_2' ]
not in the database yet: [ 'memory:acme:lead_6' ]

It reports and changes nothing: deleting the rows, or marking them, is your decision. It lists the whole collection, so it costs as much as a full sync. Run it less often than the sync. It covers campaigns, leads and senders.

What each vendor can serve

The same sync, run on the in-memory adapter limited to what each vendor’s adapter declares. This shows which calls a vendor’s adapter would refuse. It does not show what that vendor’s API returns.

import { OutboundError } from "outbound-sdk";

import { asVendor, VENDORS } from "../_shared/as-vendor.ts";
import type { Vendor } from "../_shared/as-vendor.ts";
import { openDatabase } from "./database.ts";
import { seed } from "./seed.ts";
import { syncAll, syncing } from "./sync-all.ts";

const syncAs = async (vendor: Vendor): Promise<string> => {
  const db = openDatabase();
  const outbound = syncing(asVendor(vendor, seed, "acme"), db);
  const { capabilities } = outbound;
  const filters = `leads.since ${capabilities["leads.since"]}, replies.since ${capabilities["replies.since"]}`;

  try {
    const results = await syncAll(outbound, db);
    const records = results.map((result) => result.records).join(" + ");

    return `${vendor}: ok, ${records} records (${filters})`;
  } catch (error) {
    if (!OutboundError.is(error)) {
      throw error;
    }

    return `${vendor}: ${error.code}: ${error.message} (${filters})`;
  }
};

const lines = await Promise.all(VENDORS.map(syncAs));

console.log(lines.join("\n"));

instantly: ok, 2 + 5 + 2 records (leads.since client, replies.since server)
smartlead: ok, 2 + 5 + 2 records (leads.since client, replies.since client)
heyreach: Unsupported: Listing leads is not available: the heyreach adapter does not support it (leads.list) (leads.since server, replies.since server)

• HeyReach is refused because it has no list of every lead: leads exist only inside a campaign or a list. Sync leads once per campaign by passing campaignId. Each campaign then has its own checkpoint.
• leads.since is client on Instantly and Smartlead. Their APIs cannot filter leads by date, so the SDK reads every lead and drops the older ones before your onPage sees them. The second run hands you only what is new, and still costs as many vendor requests as the first. On Smartlead the same is true of replies and messages.
• server means the vendor applies the filter, and a later run costs requests in proportion to what changed.

What it costs in vendor requests

A run is one listing per collection: the number of pages, times that vendor’s cost per page. outbound.estimate({ action: "leads.list", records }) gives the expected request count before you make the call, where the adapter declares a cost. Where a date filter is client, estimate with the total number of records, not the number that changed.

When it cannot answer

import { OutboundError } from "outbound-sdk";
import type { Campaign, Lead, Reply } from "outbound-sdk";
import { memory } from "outbound-sdk/memory";

import {
  openDatabase,
  upsertCampaigns,
  upsertLeads,
  upsertReplies,
} from "./database.ts";
import { seed } from "./seed.ts";
import { syncing } from "./sync-all.ts";

const db = openDatabase();

const outbound = syncing(
  memory({ data: seed, pageSize: 2, workspaceId: "acme" }),
  db
);

// 1. `maxPages` stops a run early. It is not a failure: the result says the
//    run is incomplete, and the next run continues from the checkpoint.
const capped = await outbound.sync.run<Lead>("leads", {
  maxPages: 1,
  onPage: (leads) => upsertLeads(db, leads),
});

console.log(`capped: complete ${capped.complete}, ${capped.records} records`);

const rest = await outbound.sync.run<Lead>("leads", {
  onPage: (leads) => upsertLeads(db, leads),
});

console.log(`next run: resumed ${rest.resumed}, ${rest.records} records`);

// 2. A collection the adapter cannot list is refused before any request.
const noReplies = syncing(
  memory({ capabilities: { "replies.list": false }, data: seed }),
  db
);

try {
  await noReplies.sync.run<Reply>("replies", {
    onPage: (replies) => upsertReplies(db, replies),
  });
} catch (error) {
  if (OutboundError.is(error)) {
    console.log("no replies:", error.code, "-", error.message);
  }
}

// 3. A failed vendor call ends the run with the SDK's error. The checkpoint
//    stays where the last written page left it.
const rejected = syncing(
  {
    ...memory({ data: seed, workspaceId: "acme" }),
    listCampaigns: () =>
      Promise.reject(
        new OutboundError("Unauthorized", "The key was rejected", {
          status: 401,
        })
      ),
  },
  db
);

try {
  await rejected.sync.run<Campaign>("campaigns", {
    onPage: (campaigns) => upsertCampaigns(db, campaigns),
  });
} catch (error) {
  if (OutboundError.is(error)) {
    console.log("bad key:", error.code, "permanent", error.permanent);
  }
}

// 4. `reset` forgets a checkpoint, so the next run reads everything again.
await outbound.sync.reset("leads");

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

console.log(
  `after reset: ${again.records} records, lower bound ${again.since ?? "none"}`
);

capped: complete false, 2 records
next run: resumed true, 3 records
no replies: Unsupported - Listing replies is not available: the memory adapter does not support it (replies.list)
bad key: Unauthorized permanent true
after reset: 5 records, lower bound none

Limits

• On Instantly a reply carries the lead’s address but no lead id, so reply.leadRef is null and the lead_ref column in this guide’s schema stays empty there. Join replies to leads on the address. The in-memory adapter fills leadRef, which hides this.
• Delivery is at least once. A page can arrive twice; it never arrives zero times. Writes must be idempotent.
• A checkpoint belongs to one provider, workspace, collection and campaign. Syncing with and without campaignId keeps two separate checkpoints.
• A dated run finds new records, not changed ones. See the note on lead status above.
• Deletions need reconcile.deletions(). It does not cover replies or messages.
• The plugin does not lock. Two concurrent runs of one collection overwrite each other’s checkpoint.
• The numbers are only as good as the vendor’s. On an unverified adapter, compare a synced table with the vendor’s own app before building on it.