Salesforce relay

// TruAgents "Send webhook" -> Salesforce Lead relay. Node 18+, no dependencies.
//
// Required environment:
//   RELAY_TOKEN       dedicated token TruAgents sends in the X-Relay-Token header
//   SF_LOGIN_URL      your My Domain URL, e.g. https://acme.my.salesforce.com
//   SF_CLIENT_ID      consumer key of the app with the Client Credentials flow
//   SF_CLIENT_SECRET  consumer secret of that app
// Optional:
//   SF_API_VERSION    default v62.0
//   PORT              default 8080

import http from "node:http";
import { timingSafeEqual } from "node:crypto";

function required(name) {
  const value = process.env[name];
  if (!value) throw new Error(`missing env ${name}`);
  return value;
}

const RELAY_TOKEN = Buffer.from(required("RELAY_TOKEN"));
const SF_LOGIN_URL = required("SF_LOGIN_URL").replace(/\/$/, "");
const SF_CLIENT_ID = required("SF_CLIENT_ID");
const SF_CLIENT_SECRET = required("SF_CLIENT_SECRET");
const SF_API_VERSION = process.env.SF_API_VERSION || "v62.0";
const PORT = Number(process.env.PORT || 8080);
const MAX_BODY_BYTES = 64 * 1024;

let cachedToken = null;

async function salesforceToken() {
  if (cachedToken) return cachedToken;
  const res = await fetch(`${SF_LOGIN_URL}/services/oauth2/token`, {
    method: "POST",
    headers: { "Content-Type": "application/x-www-form-urlencoded" },
    body: new URLSearchParams({
      grant_type: "client_credentials",
      client_id: SF_CLIENT_ID,
      client_secret: SF_CLIENT_SECRET,
    }),
  });
  if (!res.ok) throw new Error(`salesforce auth failed: ${res.status} ${await res.text()}`);
  const body = await res.json();
  cachedToken = { accessToken: body.access_token, instanceUrl: body.instance_url };
  return cachedToken;
}

function tokenMatches(header) {
  const given = Buffer.from(header || "");
  return given.length === RELAY_TOKEN.length && timingSafeEqual(given, RELAY_TOKEN);
}

// A Lead requires LastName and Company. TruAgents sends one full name and no
// company, so split the name and take the company from the email domain.
export function toSalesforceLead(event) {
  const { workflow, contact, communication } = event.data;
  const parts = (contact.name || "").trim().split(/\s+/).filter(Boolean);
  const lastName = parts.pop() || contact.email || "Unknown";
  const domain = (contact.email || "").split("@")[1] || "";
  const company = domain ? domain.split(".")[0] : "Unknown";

  return {
    FirstName: parts.join(" "),
    LastName: lastName,
    Company: company,
    Email: contact.email || null,
    Phone: contact.phone || null,
    LeadSource: "TruAgents",
    Rating: "Hot",
    Description: [
      `Hot lead detected by TruAgents workflow "${workflow.name}".`,
      `Channel: ${communication.channel}`,
      `Subject: ${communication.subject || "(none)"}`,
      `Message: ${communication.preview || ""}`,
      communication._links?.self ? `Conversation: ${communication._links.self}` : "",
      workflow._links?.self ? `Workflow run: ${workflow._links.self}` : "",
    ]
      .filter(Boolean)
      .join("\n"),
  };
}

// PATCH on the external id field creates the Lead the first time (201) and
// updates the same Lead after that (204), so retries never duplicate it.
async function upsertLead(event, isRetry = false) {
  const { accessToken, instanceUrl } = await salesforceToken();
  const contactId = encodeURIComponent(event.data.contact.id);
  const url = `${instanceUrl}/services/data/${SF_API_VERSION}/sobjects/Lead/TruAgents_Contact_Id__c/${contactId}`;
  const res = await fetch(url, {
    method: "PATCH",
    headers: { Authorization: `Bearer ${accessToken}`, "Content-Type": "application/json" },
    body: JSON.stringify(toSalesforceLead(event)),
  });
  // Salesforce sessions expire; log in again once before giving up.
  if (res.status === 401 && !isRetry) {
    cachedToken = null;
    return upsertLead(event, true);
  }
  return { status: res.status, body: await res.text() };
}

function reply(res, status, body) {
  res.writeHead(status, { "Content-Type": "application/json" });
  res.end(JSON.stringify(body));
}

async function readJson(req) {
  const chunks = [];
  let size = 0;
  for await (const chunk of req) {
    size += chunk.length;
    if (size > MAX_BODY_BYTES) throw new Error("body too large");
    chunks.push(chunk);
  }
  return JSON.parse(Buffer.concat(chunks).toString("utf8"));
}

http
  .createServer(async (req, res) => {
    if (req.method !== "POST") return reply(res, 405, { error: "POST only" });
    if (!tokenMatches(req.headers["x-relay-token"])) return reply(res, 401, { error: "bad token" });

    let event;
    try {
      event = await readJson(req);
    } catch {
      return reply(res, 400, { error: "invalid body" });
    }
    if (event.event !== "workflow.step.webhook_sent" || !event.data?.contact?.id) {
      return reply(res, 400, { error: "unexpected payload" });
    }

    try {
      const sf = await upsertLead(event);
      console.log(`event=${event.id} run=${event.data.workflow.run_id} salesforce=${sf.status}`);
      if (sf.status >= 200 && sf.status < 300) return reply(res, 200, { ok: true });
      // 503 asks TruAgents to retry (Salesforce outage, expired login, or API limit).
      // 400 tells TruAgents to stop: Salesforce rejected the data itself.
      const rateLimited =
        sf.status === 429 || (sf.status === 403 && sf.body.includes("REQUEST_LIMIT_EXCEEDED"));
      if (sf.status >= 500 || sf.status === 401 || rateLimited) {
        return reply(res, 503, { error: "salesforce unavailable" });
      }
      console.error(`event=${event.id} salesforce rejected lead: ${sf.body}`);
      return reply(res, 400, { error: "salesforce rejected lead" });
    } catch (err) {
      console.error(`event=${event.id}`, err);
      return reply(res, 503, { error: "relay error" });
    }
  })
  .listen(PORT, () => console.log(`salesforce relay listening on :${PORT}`));