@render-lab/tasks-salesforce

⚠️ Experimental: proof of concept. This package is part of the Render Tasks POC and is published for testing only. It is not fully tested or production ready. Task names, inputs, outputs, and behavior can change or break in any release. Pin exact versions and expect breaking changes.

Durable Salesforce REST and Bulk API 2.0 tasks for Render Workflows.

This pack talks to the Salesforce REST and Bulk API 2.0 over HTTP directly rather than through a helper library. A helper would retry, poll, and paginate internally, hiding each attempt from Render Workflows. Direct HTTP keeps every poll, result page, and retry as one observable durable run, with the baked-in retry policy as the single source of truth (vendor retries are off). All calls are pinned to API version v67.0 under /services/data/v67.0.

Install

pnpm add @render-lab/tasks-salesforce

import {
  getRecord,
  queryRecords,
  createRecord,
  updateRecord,
  upsertRecord,
  deleteRecord,
  submitBulkIngest,
  submitBulkQuery,
  getBulkJob,
  awaitBulkJob,
  getBulkResults,
  cancelBulkJob,
} from "@render-lab/tasks-salesforce";

Importing any symbol registers all 12 salesforce.* tasks with the Workflows TaskRegistry.

Tasks

| Task | Input | Output | Retry | | --- | --- | --- | --- | | salesforce.getRecord | GetRecordInput | SalesforceRecord | read (3× backoff) | | salesforce.queryRecords | QueryRecordsInput | QueryRecordsResult | read (3× backoff) | | salesforce.createRecord | CreateRecordInput | RecordMutationResult | none (0 retries) | | salesforce.updateRecord | UpdateRecordInput | UpdateRecordResult | safe-write (3× backoff) | | salesforce.upsertRecord | UpsertRecordInput | RecordMutationResult | safe-write (3× backoff) | | salesforce.deleteRecord | DeleteRecordInput | DeleteRecordResult | safe-write (3× backoff) | | salesforce.submitBulkIngest | SubmitBulkIngestInput | BulkJobRef | none (0 retries) | | salesforce.submitBulkQuery | SubmitBulkQueryInput | BulkJobRef | none (0 retries) | | salesforce.getBulkJob | GetBulkJobInput | BulkJob | read (3× backoff) | | salesforce.awaitBulkJob | AwaitBulkJobInput | BulkJob | await (poll every 15s up to 1h) | | salesforce.getBulkResults | GetBulkResultsInput | BulkResults | read (3× backoff) | | salesforce.cancelBulkJob | CancelBulkJobInput | CancelBulkJobResult | safe-write (3× backoff) |

Every task returns a JSON-serializable DTO — never a raw Salesforce response. Records are flattened: attributes and Id are stripped and the remaining JSON lands under fields, with object taken from attributes.type.

Why creates and bulk submissions have zero retries

createRecord, submitBulkIngest, and submitBulkQuery have no stable idempotency key. A retried attempt would insert a second record or spawn a second bulk job rather than resume the first, so they run with maxRetries: 0. If one fails, resume by hand: query for the record, or inspect/abort the created job with its returned id. updateRecord, deleteRecord, and cancelBulkJob are convergent on a stable ID (and the port normalizes an already-gone 404), and upsertRecord keys on an external ID, so those retry safely.

Environment contract

Both variables are required and read lazily on the first task call (never at import), so pulling in this pack for one task never forces another task's secret. The access token is checked before the instance URL, so a caller missing both sees the auth error first.

| Variable | Required | Purpose | | --- | --- | --- | | SALESFORCE_INSTANCE_URL | yes | Org instance URL, e.g. https://acme.my.salesforce.com (a trailing slash is trimmed) | | SALESFORCE_ACCESS_TOKEN | yes | OAuth bearer token sent as Authorization: Bearer <token> |

Replay-safe upserts

Prefer upsertRecord over createRecord in workflows. Keyed on an external ID field, an upsert converges on the same record no matter how many times it replays, so it retries safely where a create cannot. The port maps the HTTP status: 201 means the record was created, 204 means it was updated — that flag rides back on RecordMutationResult.created.

const result = await upsertRecord({
  object: "Account",
  externalIdField: "ExternalId__c",
  externalId: "acme-42",
  fields: { Name: "Acme", Industry: "Tech" },
});
// result.created === true when inserted, false when an existing record was updated

SOQL pagination

queryRecords takes exactly one of soql (first page) or nextRecordsUrl (later pages). Salesforce returns an opaque relative nextRecordsUrl while done is false; follow it until it is null:

let page = await queryRecords({ soql: "SELECT Id, Name FROM Account" });
handle(page.records);
while (!page.done && page.nextRecordsUrl) {
  page = await queryRecords({ nextRecordsUrl: page.nextRecordsUrl });
  handle(page.records);
}

Bulk ingest lifecycle

submitBulkIngest performs the full create → upload CSV → close (state: "UploadComplete") sequence in one task and returns a BulkJobRef. awaitBulkJob blocks until the job reaches a terminal state, then you fetch each result category and, if needed, abort:

const ref = await submitBulkIngest({
  object: "Contact",
  operation: "upsert",
  externalIdFieldName: "ExternalId__c", // required only for upsert
  csv: "ExternalId__c,LastName\nc-1,Smith\n",
});

// Blocks (poll-by-throwing) until JobComplete; throws terminally on Aborted/Failed.
const job = await awaitBulkJob({ jobId: ref.id, kind: "ingest" });

for (const resultType of ["successful", "failed", "unprocessed"] as const) {
  const results = await getBulkResults({ jobId: ref.id, kind: "ingest", resultType });
  handleCsv(results.csv);
}

// Recovery: if submitBulkIngest throws after creating the job, its message
// carries the job ID so you can inspect or abort it:
await cancelBulkJob({ jobId: ref.id, kind: "ingest" });

submitBulkIngest rejects a UTF-8 CSV payload larger than 4 MB before any network I/O — split it into smaller batches.

Bulk query lifecycle

submitBulkQuery submits the job (operation defaults to "query"; use "queryAll" to include archived/deleted rows). Await it, then fetch bounded CSV pages using the opaque locator:

const ref = await submitBulkQuery({ soql: "SELECT Id, Name FROM Account" });
await awaitBulkJob({ jobId: ref.id, kind: "query" });

let locator: string | undefined = undefined;
do {
  const results = await getBulkResults({
    jobId: ref.id,
    kind: "query",
    resultType: "query",
    locator,
    maxRecords: 10_000,
  });
  handleCsv(results.csv);
  locator = results.nextLocator ?? undefined;
} while (locator);

Payload limits

Workflow task inputs and outputs must stay under a 4 MB payload ceiling. getBulkResults defaults maxRecords to 10,000 and maxBytes to 3,500,000, and rejects any page that exceeds maxBytes with guidance to lower maxRecords. Page through large result sets with the returned nextLocator rather than raising the limits.

No streaming or webhooks

This pack exposes REST and Bulk API 2.0 only. Salesforce Change Data Capture and Platform Events use streaming APIs (CometD / Pub/Sub) that do not fit the request/response task model, so there is no ./webhooks subpath and no generic webhook adapter here.

Testing

pnpm -C packages/tasks-salesforce test        # Tier 1: hermetic unit tests
pnpm -C packages/tasks-salesforce build
pnpm -C packages/tasks-salesforce typecheck

Tier 2 live coverage is opt-in and read-only. Set RUN_LIVE=1, SALESFORCE_INSTANCE_URL, SALESFORCE_ACCESS_TOKEN, and SALESFORCE_LIVE_RECORD_ID (optionally SALESFORCE_LIVE_OBJECT, defaulting to Account):

RUN_LIVE=1 pnpm -C packages/tasks-salesforce test:live