Automate the golden-record pipeline

This is the code half of patterns/golden-record-rag-pipeline-for-revenue.md. It uses the Connect API with the CLI's org login (sf org login web or sf org login jwt), so it needs no token exchange. The design rules are in reference/data360/automation-and-data-kits.md: capture and replay, idempotence, credit gates and dependency order.

Layout

d360/
  data-streams/Account_Home.json  Contact_Home.json  Order_Home.json  OrderItem_Home.json  Product2_Home.json
  data-model-object-mappings/<mapping>.json      # Order → Sales Order, OrderItem → Sales Order Product
  identity-resolutions/Revenue_People.json  Revenue_Accounts.json
  calculated-insights/Revenue_Rollup__cio.json
  data-transforms/Account_Narrative.json         # writes the narrative text DMO
  search-index/Account_Narrative_Index.json
  data-graphs/Revenue_Golden_Record.json
  verify.json
scripts/d360-pipeline.ts

Data streams can be written by hand from Salesforce's documented body (reference/data360/streams-and-mapping.md). Capture everything else from the development org where you first built it in the UI.

The orchestrator

// scripts/d360-pipeline.ts — build the golden-record RAG pipeline from committed JSON definitions.
// Usage: npx tsx scripts/d360-pipeline.ts <capture|apply|run|wait|verify> --org <alias> [options]
import { Org, type Connection } from "@salesforce/core";
import { mkdirSync, readFileSync, readdirSync, writeFileSync, existsSync } from "node:fs";
import { join } from "node:path";

type Json = Record<string, unknown>;
const API = `/services/data/v${process.env.SF_API ?? "66.0"}/ssot`;
const ROOT = "d360";
// Dependency order. Each folder holds one JSON definition per component, named <name>.json.
const STAGES = ["data-streams", "data-model-object-mappings", "identity-resolutions",
  "calculated-insights", "data-transforms", "search-index", "data-graphs"] as const;
type Stage = (typeof STAGES)[number];
const SERVER_OWNED = /^(id|url|status|runtimeStatus|lastRun.*|matchedRate|created.*|lastModified.*|totalRecords)$/;
const args = process.argv.slice(2);
const flag = (n: string) => { const i = args.indexOf(`--${n}`); return i >= 0 ? args[i + 1] : undefined; };
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

async function call<T = Json>(conn: Connection, method: string, path: string, body?: unknown): Promise<T> {
  return conn.request<T>({ method: method as "GET", url: `${API}${path}`,
    body: body === undefined ? undefined : JSON.stringify(body), headers: { "Content-Type": "application/json" } });
}
const nameOf = (r: Json) => String(r.name ?? r.apiName ?? r.developerName ?? r.label ?? "");
const firstArray = (r: Json): Json[] => (Object.values(r).find(Array.isArray) as Json[] | undefined) ?? [];
const strip = (r: Json): Json => Object.fromEntries(Object.entries(r).filter(([k]) => !SERVER_OWNED.test(k)));

async function findByName(conn: Connection, stage: Stage, name: string): Promise<Json | undefined> {
  const list = firstArray(await call(conn, "GET", `/${stage}`));
  return list.find((r) => nameOf(r) === name);
}

async function capture(conn: Connection, stage: Stage, name: string): Promise<void> {
  const hit = await findByName(conn, stage, name);
  if (!hit) throw new Error(`${stage}/${name} not found`);
  const full = await call(conn, "GET", `/${stage}/${encodeURIComponent(String(hit.id ?? name))}`);
  mkdirSync(join(ROOT, stage), { recursive: true });
  writeFileSync(join(ROOT, stage, `${name}.json`), JSON.stringify(strip(full), null, 2) + "\n");
  console.log(`captured ${stage}/${name}`);
}

async function apply(conn: Connection): Promise<void> {
  for (const stage of STAGES) {
    const dir = join(ROOT, stage);
    if (!existsSync(dir)) continue;
    for (const file of readdirSync(dir).filter((f) => f.endsWith(".json")).sort()) {
      const def = JSON.parse(readFileSync(join(dir, file), "utf8")) as Json;
      if (await findByName(conn, stage, nameOf(def))) { console.log(`exists  ${stage}/${nameOf(def)}`); continue; }
      await call(conn, "POST", `/${stage}`, def);
      console.log(`created ${stage}/${nameOf(def)}`);
    }
  }
}

// Credit-consuming runs, in dependency order. Refuses unless --allow-billable is passed.
async function run(conn: Connection): Promise<void> {
  if (!args.includes("--allow-billable")) throw new Error("runs consume Data 360 credits: pass --allow-billable");
  const each = async (stage: Stage, action: (r: Json) => string) => {
    for (const r of firstArray(await call(conn, "GET", `/${stage}`))) {
      if (!existsSync(join(ROOT, stage, `${nameOf(r)}.json`))) continue; // only components this pipeline owns
      await call(conn, "POST", action(r));
      console.log(`started ${stage}/${nameOf(r)}`);
    }
  };
  await each("identity-resolutions", (r) => `/identity-resolutions/${String(r.id)}/actions/run-now`);
  await each("calculated-insights", (r) => `/calculated-insights/${nameOf(r)}/actions/run`);
  await each("data-transforms", (r) => `/data-transforms/${nameOf(r)}/actions/run`);
  await each("data-graphs", (r) => `/data-graphs/${nameOf(r)}/actions/refresh`);
}

// Poll until owned identity resolutions report success and search indexes report Ready.
async function wait(conn: Connection, minutes: number): Promise<void> {
  const until = Date.now() + minutes * 60_000;
  for (;;) {
    const pending: string[] = [];
    for (const stage of ["identity-resolutions", "search-index"] as const) {
      for (const r of firstArray(await call(conn, "GET", `/${stage}`))) {
        if (!existsSync(join(ROOT, stage, `${nameOf(r)}.json`))) continue;
        const state = String(r.lastRunStatus ?? r.runtimeStatus ?? r.status ?? "");
        if (!/^(success|succeeded|ready|active)$/i.test(state)) pending.push(`${stage}/${nameOf(r)}=${state}`);
      }
    }
    if (pending.length === 0) return console.log("all ready");
    if (Date.now() > until) throw new Error(`still waiting: ${pending.join(", ")}`);
    console.log(`waiting: ${pending.join(", ")}`);
    await sleep(60_000);
  }
}

// d360/verify.json: [{ "name": "...", "sql": "SELECT COUNT(*) FROM ...", "min": 1 }]
async function verify(conn: Connection): Promise<void> {
  const checks = JSON.parse(readFileSync(join(ROOT, "verify.json"), "utf8")) as { name: string; sql: string; min: number }[];
  let failed = 0;
  for (const c of checks) {
    const res = await call<{ data: unknown[][] }>(conn, "POST", "/query-sql", { sql: c.sql });
    const value = Number(res.data?.[0]?.[0] ?? 0);
    const ok = value >= c.min;
    if (!ok) failed++;
    console.log(`${ok ? "ok  " : "FAIL"} ${c.name}: ${value} (min ${c.min})`);
  }
  if (failed) process.exit(1);
}

async function main(): Promise<void> {
  const org = await Org.create({ aliasOrUsername: flag("org") });
  const conn = org.getConnection();
  switch (args[0]) {
    case "capture": return capture(conn, flag("stage") as Stage, String(flag("name")));
    case "apply": return apply(conn);
    case "run": return run(conn);
    case "wait": return wait(conn, Number(flag("minutes") ?? 45));
    case "verify": return verify(conn);
    default: throw new Error("command: capture | apply | run | wait | verify");
  }
}
main().catch((e: unknown) => { console.error(e instanceof Error ? e.message : e); process.exit(1); });

Run it

# once, in the development org where the pipeline was built in the UI
npx tsx scripts/d360-pipeline.ts capture --org dev --stage identity-resolutions --name Revenue_People
npx tsx scripts/d360-pipeline.ts capture --org dev --stage calculated-insights --name Revenue_Rollup__cio
# every environment (CI): create, run, wait, run again (insights follow the new golden records), verify
npx tsx scripts/d360-pipeline.ts apply --org target
npx tsx scripts/d360-pipeline.ts run --org target --allow-billable
npx tsx scripts/d360-pipeline.ts wait --org target --minutes 45
npx tsx scripts/d360-pipeline.ts run --org target --allow-billable
npx tsx scripts/d360-pipeline.ts verify --org target

d360/verify.json, with one check per stage:

[
  { "name": "orders landed", "sql": "SELECT COUNT(*) FROM ssot__SalesOrder__dlm", "min": 1 },
  { "name": "golden people", "sql": "SELECT COUNT(*) FROM UnifiedIndividual__dlm", "min": 1 },
  { "name": "revenue rollup", "sql": "SELECT COUNT(*) FROM Revenue_Rollup__cio", "min": 1 },
  { "name": "narrative rows", "sql": "SELECT COUNT(*) FROM Account_Narrative__dlm", "min": 1 }
]

What to check before trusting it

• List shapes. The script takes the first array in each list response, and matches on name, apiName, developerName or label. That's how the community plugin and sf-skills read these lists (for example dataModelObject[] and identityResolutions[]). If a list needs filters (mappings may), add them in findByName.
• Captured JSON may carry org-specific values, such as connection names, data space names and the org ID in CRM bundles. Keep those as placeholders and substitute them per org before apply.
• apply creates but doesn't update. For a changed definition, PATCH the component or delete and re-create it deliberately; identity resolution re-runs consume credits and daily run quota.
• Run limits. run starts every owned identity resolution ruleset. On-demand runs are capped per day (reference/data360/identity-resolution.md).
• Promotion between orgs can use data kits instead of replay (reference/data360/automation-and-data-kits.md). Use one approach per component, not both.

Wiring the result into an agent: guides/data360/ground-a-revenue-agent.md.