Custom data implementation planner

Validate the source

-- Run this before you build the stream.

-- 1. Is the primary key actually unique, and never null?
SELECT COUNT(Id) total,
       COUNT_DISTINCT(Record_Key__c) distinct_keys
FROM Example_Object__c

-- If total and distinct_keys differ, your primary key is not one.
-- If they agree here but not in production, you tested in a sandbox
-- against a subset. Check production.

-- 2. How much of this is actually recent?
SELECT CALENDAR_YEAR(Occurred_At__c) yr, COUNT(Id)
FROM Example_Object__c
GROUP BY CALENDAR_YEAR(Occurred_At__c)
ORDER BY CALENDAR_YEAR(Occurred_At__c) DESC

-- Ingest a retention window rather than the whole history.
-- A validation query that runs longer than 2 minutes fails and retries,
-- so an object this cannot summarise quickly is one to filter down
-- before you point a stream at it.

Up to 30 objects can be selected per creation pass, all fields are selected by default, and a validation query that runs longer than 2 minutes fails and retries. Do not change the CRM object API name after the stream exists.

Both queries here are ours. They exist because a primary key that is not unique and a table that is mostly ancient are the two things we see sink CRM connector streams, and both are visible before you build anything.

Check field access

# Run this AS THE INTEGRATION USER, not as yourself.

sf org login web --alias prod

# The fields that user can actually read:
sf sobject describe --sobject Example_Object__c --target-org prod \
  --json | jq -r '.result.fields[] | select(.accessible==true) | .name'

# Diff that against the fields you planned to map:
#   Record_Key__c
#   Party__c
#   Occurred_At__c
#   Amount__c

# Anything missing is a field-level security problem, not a Data 360
# problem. Fix it in the permission set before you continue.
#
# And if this object is a big object, stop: the CRM connector does not
# support them. It has to come through the Ingestion API instead.

The CRM connector ingests from the connected org and sends back only data actions, and big objects are not supported at all.

Running the describe as the integration user rather than as yourself is the whole point of this tab. In our experience a field the integration user cannot read is the most common silent failure on this path: the field never appears in the stream and no error is raised anywhere.

Refresh and derived fields

# Not code. The two settings that decide whether this stream stays true.
#
# Incremental refresh runs every 10 minutes, starting after a full
# refresh. Periodic full refresh is disabled by default in new data
# streams, and its interval is configurable once you enable it.
#
# Formulas are refreshed during a full or incremental refresh and are
# recalculated if a record has changed. A formula value changing does
# not update SystemModstamp or LastModifiedDate.
#
# Our reading of those two together: a formula whose inputs moved on
# another record leaves nothing for an incremental refresh to detect,
# so with periodic full refresh off by default the stale value can sit
# there indefinitely.
#
# No formula or roll-up fields in this set, so nothing here is exposed
# to that.
#
# Including a batch-ingested formula field in a CRM data stream also
# changes the processing mode from streaming to batch, so a derived
# field can cost you low latency as well as freshness.

Incremental refresh runs every 10 minutes starting after a full refresh, and periodic full refresh is disabled by default in new data streams. Formulas are refreshed during a full or incremental refresh and are recalculated if a record has changed, but a formula value changing updates neither SystemModstamp nor LastModifiedDate. Including a batch-ingested formula field in a CRM data stream also changes the processing mode from streaming to batch.

The conclusion below is our reading of those four facts put together rather than a sentence Salesforce publishes.

Schema (YAML)

# Uploaded once, when you create the Ingestion API source.
# Field names here are THE CONTRACT: a field cannot be removed once
# added and an existing field's type cannot change, so argue about
# them now rather than after the first load.

openapi: 3.0.3
info:
  title: Example_Object
  version: 1.0.0
components:
  schemas:
    Example_Object:
      type: object
      required:
        - record_key
      properties:
        record_key:
          type: string
        party_email:
          type: string
        occurred_at:
          type: string
          format: date-time
        amount:
          type: number

The schema is an OpenAPI 3.0.x file with a .yml or .yaml extension. No nested objects, at most 1,000 fields per object, object names at most 79 characters and field names at most 39. An object cannot be deleted from a schema once uploaded, a field cannot be removed once added, an existing field's data type cannot change, and an object intended for the engagement category must contain a datetime field.

Two things below are ours rather than quoted. The schema file accepts nine field types, boolean, date, datetime, email, number, phone, percent, text and URL, which is on the schema requirements page but is not among the facts Phase 0 recorded in a form we can cite. And how each of those names is spelled as an OpenAPI type and format pair is OpenAPI's own convention rather than something the Salesforce page states. Check both against the schema requirements page before you upload.

Authenticate

#!/usr/bin/env bash
# Two hops. Everyone forgets the second one.
#
# The connected app needs OAuth enabled with the cdp_ingest_api and api
# scopes, plus refresh_token or offline_access.

# Hop 1: a Salesforce access token. The JWT bearer flow is the one
# documented for a server to server integration; the endpoint path and
# the grant type string below are NOT documented in anything recorded
# for this tool, so both are variables. Read them off the Get Started
# page linked under Sources.
ACCESS_TOKEN=$(curl -s "$TOKEN_URL" \
  -d grant_type="$JWT_GRANT" \
  -d assertion="$SIGNED_JWT" | jq -r .access_token)

# Hop 2: exchange it at your instance_url for a Data 360 token.
#
# Same again, and one step further: the parameter NAMES below are ours
# too. What was recorded is that the first token is exchanged at the
# instance_url and that the response's instance_url is the tenant
# endpoint. The path, the grant type and subject_token were not, so
# check all five strings on that page before you run this. A wrong path
# pasted into a script is worse than a look-up.
DC=$(curl -s "$TOKEN_EXCHANGE_URL" \
  -d grant_type="$TOKEN_EXCHANGE_GRANT" \
  -d subject_token="$ACCESS_TOKEN" \
  -d subject_token_type=urn:ietf:params:oauth:token-type:access_token)

DC_TOKEN=$(echo "$DC" | jq -r .access_token)
DC_HOST=$(echo  "$DC" | jq -r .instance_url)

# DC_HOST is a DIFFERENT host from your org: it is the tenant specific
# Ingestion API endpoint, and every call below uses it.
#
# Token lifetime follows the connected app's session timeout policy.
# Refresh on a 401 rather than on a timer, and make sure the refresh
# is not racing your own workers.

Set up an Ingestion API connector and a connected app with OAuth enabled, with the cdp_ingest_api and api scopes plus refresh_token or offline_access. Acquire a Salesforce access token, with the OAuth 2.0 JWT bearer flow documented for server-to-server integration, then exchange it at your instance_url for a Data 360 access token. The instance_url in the exchange response is the tenant-specific Ingestion API endpoint, and token lifetime follows the connected app's session timeout policy.

Everything else in the script below is ours and needs checking against that page before you run it: the hop 1 token endpoint path, the JWT bearer grant type string, the exchange endpoint path, the exchange grant type, and the subject_token and subject_token_type parameter names. Phase 0 recorded the shape of this flow and none of those strings, so the two paths and the exchange grant type are left as variables rather than guessed, and the parameter names are written out only because a script with no parameters at all would not communicate the shape.

CSV payload

record_key,party_email,occurred_at,amount
EXAMPLE-001,[email protected],2026-09-08T14:22:00.000Z,129.99

# The header must match the schema field names exactly, case included.
# Empty means "no value". Datetimes are ISO 8601 UTC in Zulu form.
#
# Each file is at most 150 MB and a job takes at most 100 files, so
# partition a backfill by date rather than sending one enormous file.
# One job per month or per week gives you a restart point when
# something fails at eighty per cent, and something will.

Bulk CSV is UTF-8, RFC 4180, comma delimited only, and the header row must exactly match the data source object's field names including case. Empty cells become null, dates are yyyy-MM-dd and datetimes are ISO 8601 UTC in Zulu form. An update is a full replace: there are no patch semantics. A file is at most 150 MB and a job takes at most 100 of them, uploaded one at a time.

Bulk job lifecycle

#!/usr/bin/env bash
set -euo pipefail

# The documented rate limit, quoted exactly as the page phrases it:
#
#   "Number of Requests or Jobs Allowed per Hour: 20"
#
# It does not say whether that counts jobs, uploads or every request,
# so plan against the strictest reading and do not restate it.
# Concurrent jobs or requests: 5.

# 1. Open a job. "upsert" keys on the primary key in your schema, which
#    makes a replay safe. Append-only history can use "insert", but a
#    replay then duplicates: choose deliberately.
JOB=$(curl -sS -X POST "$DC_HOST/api/v1/ingest/jobs" \
  -H "Authorization: Bearer $DC_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"object":"Example_Object","sourceName":"Example_Object_Stream","operation":"upsert"}' \
  | jq -r .id)

# 2. Upload the CSV files, one per request, against the SAME job.
for f in ./out/example_object_*.csv; do
  curl -sS -X PUT "$DC_HOST/api/v1/ingest/jobs/$JOB/batches" \
    -H "Authorization: Bearer $DC_TOKEN" \
    -H "Content-Type: text/csv" \
    --data-binary @"$f"
done

# 3. Close it. Nothing is processed until you do.
curl -sS -X PATCH "$DC_HOST/api/v1/ingest/jobs/$JOB" \
  -H "Authorization: Bearer $DC_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"state":"UploadComplete"}'

# 4. Poll, then read the failure count. A job can reach JobComplete
#    having rejected rows.
curl -sS "$DC_HOST/api/v1/ingest/jobs/$JOB" \
  -H "Authorization: Bearer $DC_TOKEN" \
  | jq '{state,numberRecordsProcessed,numberRecordsFailed}'

# Open or UploadComplete jobs older than 7 days are deleted from the
# queue. After ingestion, allow a minimum of 30 seconds for caches
# before you query 4 fields' worth of anything.

A job is created with an object, a source name and an operation of upsert or delete; files are uploaded one per request with Content-Type text/csv; the job is closed by patching its state to UploadComplete or aborted by patching it to Aborted. The states are Open, UploadComplete, InProgress, JobComplete, Failed and Aborted, an aborted job is not processed and its uploaded data is deleted, and a job can complete having rejected rows, so read numberRecordsFailed. Open or UploadComplete jobs older than 7 days are deleted from the queue, 5 jobs or requests may run concurrently, and after ingestion you should allow a minimum of 30 seconds for caches before querying.

The per-hour limit is quoted below exactly as the documentation phrases it, and deliberately not restated. It does not define whether it counts jobs, uploads or all requests, and any tidier sentence we wrote would be an invention.

Schema (YAML)

# Uploaded once, when you create the Ingestion API source.
# Field names here are THE CONTRACT: a field cannot be removed once
# added and an existing field's type cannot change, so argue about
# them now rather than after the first load.

openapi: 3.0.3
info:
  title: Example_Event
  version: 1.0.0
components:
  schemas:
    Example_Event:
      type: object
      required:
        - record_key
        - occurred_at
      properties:
        record_key:
          type: string
        party_key:
          type: string
        occurred_at:
          type: string
          format: date-time
        amount:
          type: number

The schema is an OpenAPI 3.0.x file with a .yml or .yaml extension. No nested objects, at most 1,000 fields per object, object names at most 79 characters and field names at most 39. An object cannot be deleted from a schema once uploaded, a field cannot be removed once added, an existing field's data type cannot change, and an object intended for the engagement category must contain a datetime field.

Two things below are ours rather than quoted. The schema file accepts nine field types, boolean, date, datetime, email, number, phone, percent, text and URL, which is on the schema requirements page but is not among the facts Phase 0 recorded in a form we can cite. And how each of those names is spelled as an OpenAPI type and format pair is OpenAPI's own convention rather than something the Salesforce page states. Check both against the schema requirements page before you upload.

Authenticate

#!/usr/bin/env bash
# Two hops. Everyone forgets the second one.
#
# The connected app needs OAuth enabled with the cdp_ingest_api and api
# scopes, plus refresh_token or offline_access.

# Hop 1: a Salesforce access token. The JWT bearer flow is the one
# documented for a server to server integration; the endpoint path and
# the grant type string below are NOT documented in anything recorded
# for this tool, so both are variables. Read them off the Get Started
# page linked under Sources.
ACCESS_TOKEN=$(curl -s "$TOKEN_URL" \
  -d grant_type="$JWT_GRANT" \
  -d assertion="$SIGNED_JWT" | jq -r .access_token)

# Hop 2: exchange it at your instance_url for a Data 360 token.
#
# Same again, and one step further: the parameter NAMES below are ours
# too. What was recorded is that the first token is exchanged at the
# instance_url and that the response's instance_url is the tenant
# endpoint. The path, the grant type and subject_token were not, so
# check all five strings on that page before you run this. A wrong path
# pasted into a script is worse than a look-up.
DC=$(curl -s "$TOKEN_EXCHANGE_URL" \
  -d grant_type="$TOKEN_EXCHANGE_GRANT" \
  -d subject_token="$ACCESS_TOKEN" \
  -d subject_token_type=urn:ietf:params:oauth:token-type:access_token)

DC_TOKEN=$(echo "$DC" | jq -r .access_token)
DC_HOST=$(echo  "$DC" | jq -r .instance_url)

# DC_HOST is a DIFFERENT host from your org: it is the tenant specific
# Ingestion API endpoint, and every call below uses it.
#
# Token lifetime follows the connected app's session timeout policy.
# Refresh on a 401 rather than on a timer, and make sure the refresh
# is not racing your own workers.

Set up an Ingestion API connector and a connected app with OAuth enabled, with the cdp_ingest_api and api scopes plus refresh_token or offline_access. Acquire a Salesforce access token, with the OAuth 2.0 JWT bearer flow documented for server-to-server integration, then exchange it at your instance_url for a Data 360 access token. The instance_url in the exchange response is the tenant-specific Ingestion API endpoint, and token lifetime follows the connected app's session timeout policy.

Everything else in the script below is ours and needs checking against that page before you run it: the hop 1 token endpoint path, the JWT bearer grant type string, the exchange endpoint path, the exchange grant type, and the subject_token and subject_token_type parameter names. Phase 0 recorded the shape of this flow and none of those strings, so the two paths and the exchange grant type are left as variables rather than guessed, and the parameter names are written out only because a script with no parameters at all would not communicate the shape.

Send records

# One call, one set of records.

curl -sS -X POST \
  "$DC_HOST/api/v1/ingest/sources/Example_Event_Stream/Example_Event" \
  -H "Authorization: Bearer $DC_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
  "data": [
    {
      "record_key": "EXAMPLE-001",
      "party_key": "EXAMPLE-001",
      "occurred_at": "2026-09-08T14:22:00.000Z",
      "amount": "129.99"
    }
  ]
}'

# Expect 202 Accepted. That means QUEUED, not stored: processing is
# asynchronous and runs approximately every 3 minutes, and you should
# allow a minimum of 30 seconds for caches before you query.
#
# Use the schema's field names, not Salesforce API names. A payload
# using API names produces a single row of blank values and no error.
#
# While developing, validate synchronously first:
#   POST .../api/v1/ingest/sources/Example_Event_Stream/Example_Event/actions/test

Records are posted to the ingest source endpoint with a body of {"data": [records]}, using the schema's field names rather than Salesforce API names. A 202 Accepted means queued rather than stored, processing is asynchronous and runs approximately every 3 minutes, and after ingestion you should allow a minimum of 30 seconds for caches before querying. There is a synchronous validation endpoint for development that checks a payload before committing it.

Batching and retry

// Batching, upsert semantics and retry are the three things that bite.
//
// There is no records-per-request limit published: the documented
// limit is a 200 KB JSON body, so batch against BYTES rather than
// against a record count somebody guessed.

const MAX_BODY_BYTES = 200 * 1024;   // 200 KB, documented
const BACKOFF_BASE_MS = 500;         // ours, not a Salesforce limit

async function send(records) {
  let batch = [];
  for (const record of records) {
    const next = [...batch, record];
    if (Buffer.byteLength(JSON.stringify({ data: next })) > MAX_BODY_BYTES) {
      await flush(batch);
      batch = [record];
    } else {
      batch = next;
    }
  }
  if (batch.length > 0) await flush(batch);
}

async function flush(batch) {
  await withRetry(() =>
    fetch(`${DC_HOST}/api/v1/ingest/sources/Example_Event_Stream/Example_Event`, {
      method: 'POST',
      headers: {
        Authorization: `Bearer ${dcToken}`,
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({ data: batch }),
    })
  );
}

async function withRetry(fn, attempts = 5) {
  for (let a = 0; a < attempts; a++) {
    const res = await fn();
    if (res.ok) return res;
    if (res.status === 401) { await refreshToken(); continue; }
    if (res.status === 429 || res.status >= 500) {
      // 429 means reduce request frequency. There are 250 requests per
      // second across ALL object endpoints, so back off globally and
      // not just on this worker.
      await sleep(2 ** a * BACKOFF_BASE_MS + Math.random() * BACKOFF_BASE_MS);
      continue;
    }
    throw new Error(`${res.status} ${await res.text()}`);
  }
  throw new Error('gave up after retries');
}

// A record that vanished at the source is invisible here. Send a
// delete explicitly, at most 200 records a call.
// Deletes key on record_key.

The documented limits are a maximum JSON body of 200 KB per request and 250 total requests per second across all Ingestion API object endpoints, with HTTP 429 meaning reduce request frequency. A delete takes at most 200 records per call.

No records-per-request figure is documented anywhere, so this batches against the documented body size rather than against a record count. The backoff timings are ours.

CSV shape

record_key,party_email,score
EXAMPLE-001,[email protected],129.99

# The rules that matter more than they look:
#  - UTF-8, RFC 4180 quoting, header row always present
#  - timestamps ISO 8601 in UTC, never local time
#  - an empty cell means "no value", not "clear this value"
#  - column order is part of the contract once the stream exists

Everything on this tab is our convention. Salesforce documents CSV requirements for the Ingestion API's bulk path, and a cloud storage or SFTP drop is a different connector whose file conventions Phase 0 found nothing on. We are not going to borrow the bulk rules and present them as this path's.

Naming and manifest

# Partner drops fail on ambiguity about what a file MEANS, not on
# transport. Put the answer in the name.

s3://your-bucket/example_feed/
    example_feed_full_YYYYMMDD.csv      # full replace
    example_feed_delta_YYYYMMDD.csv     # incremental

# Decide these three, and write the answers down:
#  1. Is each file a FULL replace or a DELTA? A full file processed as
#     a delta double counts. A delta processed as a full silently
#     deletes everything absent from it.
#  2. What happens on a missing day? Nothing, or an empty file?
#     "Nothing" and "empty" must not mean the same thing.
#  3. Who is paged when a file does not arrive?

Everything on this tab is our convention. Salesforce documents CSV requirements for the Ingestion API's bulk path, and a cloud storage or SFTP drop is a different connector whose file conventions Phase 0 found nothing on. We are not going to borrow the bulk rules and present them as this path's.

Collection schedule

# Data 360 collects on a schedule. You do not push. Line the schedule
# up AFTER the vendor's delivery window, with slack for their retries.

# Vendor delivers 02:00 to 03:00 UTC  ->  schedule the stream at 04:00 UTC.

# Then verify the row count actually moved. A stream that runs
# successfully against a file that never landed reports success, and
# nothing downstream will tell you otherwise.

Everything on this tab is our convention. Salesforce documents CSV requirements for the Ingestion API's bulk path, and a cloud storage or SFTP drop is a different connector whose file conventions Phase 0 found nothing on. We are not going to borrow the bulk rules and present them as this path's.