Docs / Data & bulk sync

Syncing data between two systems

Hand-written. Unlike its neighbours in this folder, this page is not generated — npm run docs will not touch it. The function signatures it uses are generated, in function-reference.md.

This is the page for the job almost every integration turns out to be: keep the people, organisations, tickets or products in one system matching another one, every night or every time something changes, without anybody watching it.

It is written in five recipes. Read the first section before any of them — it is the one thing that decides whether a sync works on the test tenant and then falls over on the real one.

First: the shape that does not scale

The instinct, coming from ordinary JS, is to fetch a list and act on each item.

// Do not write this.
const people = fetch({ connection: "hr", method: "GET", path: "/people" }).body;

const results = people.map(function (person) {
  return CreatePerson({ record: person });
});

return results.length;

Every platform function works inside a callback, so this runs. That is exactly the problem: it runs on the forty people in your sandbox and it is still running at breakfast on the twelve thousand in production. Three separate things go wrong at that size.

A list from one call is only ever one page. /people answered the first hundred, or the first thousand, and the rest is behind a cursor nobody followed.

The calls happen one at a time, and nothing can stop them. A callback’s only exit is throw, which abandons the whole iteration and tells you nothing about where it got to. A plain for loop is better — it can break — but twelve thousand calls is still twelve thousand calls, and if the far end starts answering 429 the run parks and resumes once per call, for hours.

Nothing is written down. The run holds twelve thousand records in memory, and if it is interrupted at nine thousand it starts again at zero.

The answer is not a better loop. It is to stop moving the data through the script at all:

// The whole pull: every page, staged, keeping its place across a pause.
const pulled = dsFetchInto("ListPeople", {}, "run:people");

log(pulled.rows + " people staged over " + pulled.pages + " pages");

A dataset is a staging area that lives outside the run. Rows go in and out of it by name, the platform does the row work, and the run holds nothing. Everything below is built on that one idea:

Each of them takes one bounded piece of work per turn, records where it got to, and frees the worker in between. A pause — including the one a 429 causes — keeps the place. That is what makes a twelve-thousand-record sync something a run can simply do.

When you do need per-record logic

Some record work is genuinely yours: which fields go across, what counts as a duplicate, when to create and when to update. That stays in script, in the page loop:

let at = 0;
for (;;) {
  const page = dsPage("run:people", { offset: at, limit: 200 });
  if (page.length === 0) { break; }

  for (const row of page) {
    upsertBySourceId("people", row.value, {
      createFn: "CreatePerson",
      updateFn: "UpdatePerson",
    });
  }

  at = at + page.length;
  checkpoint();
}

Two things make this different from the loop at the top of the page. The rows come from a dataset a page at a time, so the run never holds the set. And checkpoint() at the end of each page writes the run’s progress down, so an interruption resumes at the page it reached instead of at the beginning.

The same handler body works for one record from a webhook and for a page of them from a nightly pull — which is recipe 1 and recipe 2 respectively.

The report

Every recipe below writes to one dataset called report, and ends by sending it. A report row names a record and a reason — never a row number:

{ "name": "Ada Lovelace", "action": "created", "outcome": "ok" }
{ "name": "Katherine Johnson", "action": "created", "outcome": "failed", "reason": "department unknown" }

The sync functions write those rows themselves. Your script only has to clear the report when it starts — otherwise last night’s rows are still in it:

dsClear("report");

and send it when it finishes:

const done = sendReport("email", { fn: "SyncReportMail" });

log(done.total + " records, " + done.failed + " need attention");

sendReport counts where the rows are and names the failures, up to a cap. It never reads the report into your script, and neither should you — see Counting where the rows are at the end.

Recipe 1 — a webhook, one record at a time

The source system calls you when something changes. This is the simplest sync and the one that keeps the two systems closest.

dsClear("report");

const person = input.event.body;

const done = upsertBySourceId("people", person, {
  createFn: "CreatePerson",
  updateFn: "UpdatePerson",
});

if (done.outcome === "failed") {
  log("could not sync " + person.displayName + ": " + done.reason);
}

return { synced: done.action, target: done.target };

upsertBySourceId asks the platform whether it has seen this source record before — that is what the id mapping is, and the platform keeps it — then calls whichever of your two functions fits, and remembers the new id when a create succeeds. Your two functions hold all the record logic; nothing in them knows what a mapping is.

A webhook does not replace a nightly sync. Deliveries are missed, systems are down, and changes made while your endpoint was unreachable are simply gone. Run recipe 2 nightly underneath this one.

Recipe 2 — a nightly delta over a REST API

The workhorse. Pull everything, compare it with last night, and act only on what changed.

dsClear("report");

// 1. Pull every page into this run's own staging area.
dsFetchInto("ListPeople", {}, "run:people");

// 2. What changed since last night? `people-baseline` is last run's copy.
const delta = stageDiff("people-baseline", "run:people", ["added", "changed", "removed"], "run:delta");

log(delta.added + " new, " + delta.changed + " changed, " + delta.removed + " gone");

// 3. Create and update — the page loop, over the rows that actually differ.
for (const kind of ["added", "changed"]) {
  let at = 0;
  for (;;) {
    const page = dsPage("run:delta." + kind, { offset: at, limit: 200 });
    if (page.length === 0) { break; }

    for (const row of page) {
      upsertBySourceId("people", row.value, {
        createFn: "CreatePerson",
        updateFn: "UpdatePerson",
      });
    }

    at = at + page.length;
    checkpoint();
  }
}

// 4. Relationships, once both ends exist. A person belongs to an
//    organisation and reports to a manager — two links, and the first one's
//    ends live in two DIFFERENT mapping namespaces.
applyLinks("run:delta.added", {
  linkFn: "SetDepartment",
  from: { namespace: "people", path: "id" },
  to: { namespace: "orgs", path: "orgId" },
});

applyLinks("run:delta.added", {
  linkFn: "SetManager",
  from: { namespace: "people", path: "id" },
  to: { namespace: "people", path: "managerId" },
});

// 5. People who left. Disabling is not deleting — see recipe 5.
let goneAt = 0;
for (;;) {
  const page = dsPage("run:delta.removed", { offset: goneAt, limit: 200 });
  if (page.length === 0) { break; }

  for (const row of page) {
    // Never mapped means never created here: there is nothing to disable,
    // and a report row for each would bury the ones that matter.
    const target = getMapping("people", row.key);
    if (target !== null) { DisablePerson({ target: target }); }
  }

  goneAt = goneAt + page.length;
  checkpoint();
}

// 6. This run's pull becomes next run's baseline.
dsClear("people-baseline");
dsRename("run:people", "people-baseline");

return sendReport("email", { fn: "SyncReportMail" });

Four things are worth pointing at.

stageDiff reads a removal from the baseline. A record that is gone is not in the fresh pull at all — the only copy left is last night’s, which is the side the platform takes it from. Getting this wrong means a disable phase that silently does nothing.

Creates come before links. You cannot set a manager who does not exist yet. applyLinks resolves both ends through the id mappings, a page at a time, and a manager who genuinely is not there yet becomes a report row rather than an error — the link gets written on the run that creates them. Note what the department link does: its two ends resolve from two different namespaces, people on one side and orgs on the other. Each side says where its mapping lives, so a link across entity types is the same one call as a link within one.

The baseline swap is last. If the run fails halfway, the baseline is still last night’s and tomorrow’s diff is still correct. Promote it only once the work is done.

Nothing costs anything when nothing changed. An unchanged person is not in any of the three staged sets, so no call goes out for them at all.

The variant: a delta token

Some APIs will tell you what changed, so you do not have to work it out:

const since = getState("people-delta-token");

const pulled = dsFetchInto("ListPeopleDelta", { token: since }, "run:changes");

// … the page loop over run:changes …

setState("people-delta-token", pulled.delta);

This is much cheaper and it is not equivalent. A delta feed tells you what changed; it does not tell you what is missing because an earlier run failed halfway, and most feeds expire their tokens. Run the full comparison above on a schedule anyway — weekly is usually enough — so the two systems cannot drift apart quietly.

Recipe 3 — a bulk CSV import

Some systems will not take ten thousand REST calls but will happily take one file. The shape is the same; the last mile is different.

dsClear("report");

dsFetchInto("ListPeople", {}, "run:people");

// 1. Pre-flight: a unique field that collides twice would fail the import
//    halfway through, with the far end deciding which half.
const clashes = dsDuplicates("run:people", "email");
if (clashes > 0) {
  // The groups themselves are staged in `<dataset>.duplicates`.
  const groups = dsPage("run:people.duplicates", { limit: 20 });
  for (const group of groups) {
    dsPut("report", "check:" + group.value.value, {
      name: group.value.value,
      action: "created",
      outcome: "failed",
      reason: group.value.count + " people share this email — fix the source first",
    });
  }
  sendReport("email", { fn: "SyncReportMail" });
  return { imported: 0, blocked: clashes };
}

// 2. Shape the rows the way the target's importer wants them.
dsMapInto("run:people", "PersonToTarget", "run:ready");

// 3. Write the files — capped, so no single upload is too big.
const files = dsToCsv("run:ready", "PeopleCsv", { maxBytes: 5000000 });

// 4. Upload each one, wait for the far end to finish, and turn the line
//    numbers it rejects back into the people who wrote those lines.
const done = importAll(files, "UploadPeople", {
  statusFn: "ImportStatus",
  pollMs: 10000,
  maxPolls: 60,
});

log(done.chunks + " files imported, " + done.rejected + " records rejected");

return sendReport("email", { fn: "SyncReportMail" });

Nodes before edges. If the file contains a manager column, the manager has to exist by the time that row is read. Either import people first and links in a second file, or sort so that managers come before their reports. The same rule as recipe 2’s ordering, in a place where you cannot see it happening.

Waiting costs nothing. importAll does not sit in a loop asking whether the import is done. It suspends the run between asks, so the worker is free and the wait is not paid for. maxPolls bounds it: a far end that never finishes is reported as such rather than hanging.

A line number is not a business fact. When the far end says “line 47 was rejected”, importAll looks up which record wrote line 47 and puts that in the report:

Katherine Johnson (imported) — duplicate email

Nobody opens the file and counts. If you need this yourself, csvLineRecord(file, line) is the same lookup.

Recipe 4 — import once, then keep up by webhook

The hybrid, and the usual answer for going live: a bulk import to load the history, then recipe 1 to stay current.

Run recipe 3 once. Then two things need care.

A bulk import usually does not fire the target’s workflows. Automations, notifications and assignment rules that would have run on a normal create are skipped — which is what you want for a ten-thousand-record backfill and is a surprise afterwards. Check what did not happen before you rely on it.

The importer knows the new ids; the platform does not. A bulk import creates records without telling you which source record became which target record, so every mapping upsertBySourceId depends on is missing — and the next webhook would create a duplicate. Bridge it by reading the target back once:

// Pull the records the import created, then record what became what.
dsFetchInto("ListTargetPeople", {}, "run:created");

let at = 0;
for (;;) {
  const page = dsPage("run:created", { offset: at, limit: 500 });
  if (page.length === 0) { break; }

  const pairs = [];
  for (const row of page) {
    if (row.value.employeeNumber) {
      pairs.push({ source: row.value.employeeNumber, target: row.value.id });
    }
  }
  setMappings("people", pairs);

  at = at + page.length;
  checkpoint();
}

This is why the import file should carry a column holding your source id. Without one there is nothing to match on, and the only way back is by name — which is not unique and will be wrong for somebody.

Recipe 5 — the hard edges

The five things that go wrong in production.

Deletion means disable

Do not delete on the strength of a record disappearing from a source. A filter change, a paused feed or a bad export makes everybody disappear at once. Disable instead, and if a record was never mapped, do nothing at all — there is nothing in the target to disable, and a report row for every one of them buries the rows that matter.

Collisions are a business decision

dsDuplicates detects; it never repairs. Which of two people owning one email address is the right one is not something the platform can decide, and picking one silently is how the wrong record gets updated for months. Report and stop, as recipe 3 does.

A record failure is not a run failure

The sync functions are fail-soft per record on purpose: ten thousand records must not die on the seven-thousandth. A refusal becomes a report row and the next record is tried. Your own mistakes are not soft — a record with no id, or a function this environment does not configure — because no later run would fix those.

A 429 is not an error

If the far end throttles you and the function says pauseOn, the platform suspends the run and resumes it when the far end said to. Progress is kept: work already done is not done twice. There is nothing to write and nothing to catch — just do not turn a bulk phase into a per-record loop, which turns one pause into thousands.

Delta tokens belong in state

setState and getState outlive the run. Store the token after the work succeeded, never before — a token stored first, on a run that then failed, skips those changes for ever.

Counting where the rows are

Never walk a dataset to add it up:

// Do not write this: the whole set through the run, to produce three numbers.
let at = 0;
const counts = {};
for (;;) {
  const page = dsPage("run:people", { offset: at, limit: 200 });
  if (page.length === 0) { break; }
  for (const row of page) {
    counts[row.value.dept] = (counts[row.value.dept] || 0) + 1;
  }
  at = at + page.length;
}

// Write this: one query, no rows moved.
const counts = dsStats("run:people", "dept");

dsCount for a total, dsStats to group, sendReport for the whole report — none of them move a row into the run.