PermuteDocs
Go to Permute

Guide

Publish a contact write sync

Send a published Data Table to Quo after reviewing a read-only preview. Publication starts delivery immediately and approves new captures every 24 hours under the same API key.

Workflow

  1. 1Publish and run a Data Table with a contacts table containing email, first_name and last_name columns. Use unique, stable, nonempty email strings as keys. Grant the API key write-sync write, source read and destination read/write access (operation sources also require access to their inputs).
  2. 2Use operations.get for the published revision and sources.get for output tables and fields. Configuration and preview validate the published output. Discover Quo destination fields and map their names to source column names. Mapping emails to the email key enables matching against Quo’s last read sync; refresh Quo before previewing when newer contacts may exist.
  3. 3Create the paused definition, capture a preview, and poll until completion or error. Review every page, including blocked and unchanged records. Quo creates missing contacts and skips unchanged matches; changed linked records and ambiguous matches remain blocked. It does not update or delete contacts.
  4. 4Use revisions.list for the version token, then revisions.publish with that token and the reviewed runId. Publication requires the same API key and a completed preview of the exact draft definition; it sends records and enables future captures. Each new operation capture follows the source’s current published revision, separately from the captured write-sync definition revision. Write syncs do not refresh upstream sources.
  5. 5Inspect every result page after delivery, even when execution is complete. Pause to stop future writes; requests already sent may finish. To resume, preview and publish again. execute sends an existing reviewed capture only while the current definition is enabled, under the capturing API key; it does not refresh source data or change the schedule.
  6. 6For a direct connector source, discover its table through sources.get and use source: { resourceId: connectorId, object: table, scope: "" } with input: { type: "source", table }. Capture uses the connector’s last read sync. Use REST or the SDK for these controls; they are not exposed through MCP.

Complete example

typescript
import { createInterface } from 'node:readline/promises';
import { PermuteClient } from '@permute/sdk';

const client = new PermuteClient({
  apiKey: process.env.PERMUTE_API_KEY!,
}).withWorkspace(process.env.PERMUTE_WORKSPACE_ID!);
const operationId = process.env.PERMUTE_CONTACTS_OPERATION_ID!;
const destinationId = process.env.PERMUTE_QUO_CONNECTOR_ID!;

const [operation, source] = await Promise.all([
  client.operations.get(operationId),
  client.sources.get(operationId),
]);
const revisionId = operation.published?.revisionId;
const table = source.tables.find((item) => item.name === 'contacts');
if (!revisionId || !table || ['email', 'first_name', 'last_name'].some(
  (name) => !table.fields?.some((field) => field.name === name),
)) throw new Error('Publish and run the expected contacts Data Table first');

const destinations = await client.writeSyncs.destinations();
const destination = destinations.items.find(
  (item) => item.id === destinationId && item.provider === 'quo',
);
const contacts = destination?.objects.find((item) => item.object === 'contacts');
if (!contacts || ['emails', 'firstName', 'lastName'].some(
  (name) => !contacts.fields.some((field) => field.name === name),
)) throw new Error('The selected Quo destination is unavailable');

const sync = await client.writeSyncs.create({
  definition: {
    source: { resourceId: operationId, object: table.name, scope: '' },
    destination: { resourceId: destinationId, object: contacts.object, scope: contacts.scope },
    input: { type: 'operation', operationId, revisionId, table: table.name },
    keyColumn: 'email',
    fields: { emails: 'email', firstName: 'first_name', lastName: 'last_name' },
  },
});
const capture = await client.writeSyncs.preview(sync.id); // No external writes.
console.log('Saved sync and preview:', sync.id, capture.id);

async function waitForRun(delivery = false) {
  const deadline = Date.now() + 10 * 60_000;
  while (Date.now() < deadline) {
    const run = await client.writeSyncs.getRun(sync.id, capture.id);
    if (run.previewError) throw new Error(run.previewError);
    if (run.execution?.error) throw new Error(run.execution.error);
    if (delivery ? run.execution?.completedAt : run.previewComplete) return run;
    await new Promise((resolve) => setTimeout(resolve, 2000));
  }
  throw new Error('Polling timed out; inspect the saved run before taking further action');
}

async function inspectAllPages(run: Awaited<ReturnType<typeof waitForRun>>) {
  let needsAttention = false;
  for (let page = 0; page < run.pageCount; page++) {
    const result = page === 0 ? run : await client.writeSyncs.getRun(sync.id, run.id, { page });
    console.log('Page', page + 1, 'of', run.pageCount);
    console.dir(result.records, { depth: null });
    needsAttention ||= result.records.some((record) =>
      record.action === 'blocked' || record.status === 'rejected' || record.status === 'unknown',
    );
  }
  return needsAttention;
}

const preview = await waitForRun();
if (await inspectAllPages(preview)) {
  throw new Error('Resolve records needing attention before publishing');
}

const terminal = createInterface({ input: process.stdin, output: process.stdout });
let approval: string;
try {
  approval = await terminal.question(
    'Review all captured values above. To send them and enable captures every 24 hours, type ' + sync.id + ': ',
  );
} finally {
  terminal.close();
}
if (approval.trim() !== sync.id) throw new Error('Sync remains paused');

const history = await client.revisions.list(sync.id);
if (!preview.definitionRevisionId || history.draft?.revisionId !== preview.definitionRevisionId) {
  throw new Error('The definition changed; capture and review a new preview');
}
await client.revisions.publish(sync.id, { expectedVersion: history.version, runId: preview.id });
const delivered = await waitForRun(true);
if (await inspectAllPages(delivered)) {
  console.error('Delivery completed with records needing attention; inspect their reasons');
}
console.log('Sync remains enabled. Pause with client.writeSyncs.pause(' + JSON.stringify(sync.id) + ').');