Migrate a homegrown webhook sender

// Before: your worker delivers and retries
for (const row of await db.claimOutboxRows(100)) {
  for (const ep of await db.endpointsFor(row.customer_id, row.event_type)) {
    const body = JSON.stringify(row.payload);
    const sig = crypto.createHmac("sha256", ep.secret).update(body).digest("hex");
    const res = await fetch(ep.url, { method: "POST", body, headers: { "content-type": "application/json", "x-signature": sig } });
    if (!res.ok) await db.scheduleRetry(row.id, ep.id, backoff(row.attempts));
  }
  await db.markDone(row.id);
}

import { WebhookRelay } from "@webhookrelay/sdk";

const relay = new WebhookRelay(); // reads RELAY_KEY and RELAY_SECRET ("Outbound: publish only" token)

for (const row of await db.claimOutboxRows(100)) {
  await relay.outbound.messages.publish(
    { consumer: row.customer_id, eventType: row.event_type, eventId: String(row.id), payload: row.payload },
    { idempotencyKey: `outbox:${row.id}` }, // a crash before markDone re-publishes safely
  );
  await db.markDone(row.id);
}

for _, row := range rows {
    _, err := api.PublishOutboundMessage(&webhookrelay.OutboundMessagePublishOptions{
        Consumer:       row.CustomerID,
        EventType:      row.EventType,
        EventID:        strconv.FormatInt(row.ID, 10),
        Payload:        row.Payload,
        IdempotencyKey: "outbox:" + strconv.FormatInt(row.ID, 10),
    })
    if err != nil {
        return err // leave the row unclaimed; the next run retries with the same key
    }
    if err := markDone(ctx, row.ID); err != nil {
        return err
    }
}

# Celery task (Python): replaces the task that POSTed to customer endpoints
import os, requests
from celery import shared_task

@shared_task(bind=True, max_retries=10, default_retry_delay=30)
def publish_webhook(self, outbox_id, customer_id, event_type, payload):
    res = requests.post(
        "https://my.webhookrelay.com/v1/outbound/messages",
        auth=(os.environ["RELAY_KEY"], os.environ["RELAY_SECRET"]),
        headers={"Idempotency-Key": f"outbox:{outbox_id}"},
        json={"consumer": customer_id, "event_type": event_type, "event_id": str(outbox_id), "payload": payload},
        timeout=10,
    )
    if res.status_code == 429 or res.status_code >= 500:
        raise self.retry()
    res.raise_for_status()  # 202 Accepted

# Sidekiq job (Ruby)
require "net/http"
require "json"

class PublishWebhookJob
  include Sidekiq::Job

  def perform(outbox_id, customer_id, event_type, payload)
    uri = URI("https://my.webhookrelay.com/v1/outbound/messages")
    req = Net::HTTP::Post.new(uri, "Content-Type" => "application/json", "Idempotency-Key" => "outbox:#{outbox_id}")
    req.basic_auth(ENV.fetch("RELAY_KEY"), ENV.fetch("RELAY_SECRET"))
    req.body = { consumer: customer_id, event_type: event_type, event_id: outbox_id.to_s, payload: payload }.to_json
    res = Net::HTTP.start(uri.host, uri.port, use_ssl: true) { |http| http.request(req) }
    raise "publish failed: #{res.code}" unless res.code == "202" # Sidekiq retries with the same key
  end
end

curl --user "$RELAY_KEY:$RELAY_SECRET" -X POST https://my.webhookrelay.com/v1/outbound/messages \
  -H 'Content-Type: application/json' -H "Idempotency-Key: outbox:$OUTBOX_ID" \
  -d "{\"consumer\":\"$CUSTOMER_ID\",\"event_type\":\"invoice.paid\",\"payload\":$PAYLOAD_JSON}"

await relay.outbound.consumers.upsert(customer.id, { name: customer.name });
const endpoint = await relay.outbound.endpoints.create(customer.id, { url, eventTypes, autoDisable: true });
// Show endpoint.secret to the customer once; read it again later with relay.outbound.endpoints.revealSecret(endpoint.id).
await relay.outbound.endpoints.update(endpoint.id, { url, eventTypes: ["invoice.paid", "invoice.voided"] });
await relay.outbound.endpoints.delete(endpoint.id);

import crypto from "node:crypto";
import { verifyWebhook } from "./verify-webhook.js"; // Standard Webhooks check from the receiving guide

export function verify(rawBody, headers) {
  if (headers["webhook-signature"]) {
    // With an imported secret, WEBHOOK_SECRET_RELAY is the same value as WEBHOOK_SECRET_OLD.
    return verifyWebhook(rawBody, headers, process.env.WEBHOOK_SECRET_RELAY);
  }
  // Your current scheme, unchanged. This example: X-Signature: <hex HMAC-SHA256 of the body>.
  const received = Buffer.from(headers["x-signature"] ?? "", "hex");
  const expected = crypto.createHmac("sha256", process.env.WEBHOOK_SECRET_OLD).update(rawBody).digest();
  return received.length === expected.length && crypto.timingSafeEqual(received, expected);
}

// migrate-diy.mts
// Env: DATABASE_URL, RELAY_KEY, RELAY_SECRET (token with "Outbound: manage"), DRY_RUN=false to write.
import { existsSync, readFileSync, writeFileSync } from "node:fs";
import pg from "pg";
import { WebhookRelay, WebhookRelayAPIError } from "@webhookrelay/sdk";

const RELAY_API = "https://my.webhookrelay.com";
const AUTH = "Basic " + Buffer.from(`${process.env.RELAY_KEY}:${process.env.RELAY_SECRET}`).toString("base64");
const DRY_RUN = process.env.DRY_RUN !== "false";
const STATE_FILE = "migration-state.json"; // your endpoint row ID -> Webhook Relay endpoint ID. Never secrets.
const ID = /^[a-zA-Z0-9][a-zA-Z0-9_.:-]{0,127}$/;

const relay = new WebhookRelay();

type PlanEndpoint = {
  sourceId: string; url: string; eventTypes: string[]; description?: string;
  headers?: Record<string, string>; rate?: number; disabledAtSource?: boolean;
  secret?: string; // in memory only: never printed, logged or written to a file
};
type Plan = {
  eventTypes: { name: string; description?: string; deprecated?: boolean }[];
  consumers: { id: string; name?: string; endpoints: PlanEndpoint[] }[];
  newSecrets: string[]; // source endpoint IDs whose secret can't be imported
  warnings: string[];
};

// A secret imports if it is 32-100 standard base64 characters (a key of at least 24 bytes),
// with or without "whsec_". The API makes the final check and never echoes the secret.
function importable(raw: string | null | undefined): string | undefined {
  const secret = raw?.trim();
  if (!secret) return undefined;
  const b64 = secret.startsWith("whsec_") ? secret.slice("whsec_".length) : secret;
  if (b64.length < 32 || b64.length > 100 || !/^[A-Za-z0-9+/]+={0,2}$/.test(b64)) return undefined;
  if (b64.replace(/=+$/, "").length % 4 === 1) return undefined;
  const bytes = Buffer.from(b64, "base64").length;
  return bytes >= 24 ? secret : undefined;
}

// ---- 1. Export from your database (read-only). Adjust to your schema. ----
async function exportPlan(): Promise<Plan> {
  const db = new pg.Client({ connectionString: process.env.DATABASE_URL });
  await db.connect();
  try {
    const { rows } = await db.query(`
      SELECT e.id::text AS id, e.customer_id::text AS customer_id, c.name AS customer_name,
             e.url, e.event_types, e.description, e.enabled, e.secret
      FROM webhook_endpoints e JOIN customers c ON c.id = e.customer_id
      WHERE e.deleted_at IS NULL
      ORDER BY e.customer_id, e.id`);
    const { rows: types } = await db.query(`SELECT DISTINCT event_type AS name FROM webhook_outbox`);

    const plan: Plan = { eventTypes: types.map((t) => ({ name: t.name })), consumers: [], newSecrets: [], warnings: [] };
    const consumers = new Map<string, Plan["consumers"][number]>();
    for (const r of rows) {
      const consumer = consumers.get(r.customer_id) ?? { id: r.customer_id, name: r.customer_name, endpoints: [] as PlanEndpoint[] };
      const eventTypes: string[] = r.event_types?.length ? r.event_types : ["*"];
      for (const t of eventTypes) if (t !== "*" && !plan.eventTypes.some((x) => x.name === t)) plan.eventTypes.push({ name: t });
      const secret = importable(r.secret);
      if (!secret) plan.newSecrets.push(r.id);
      consumer.endpoints.push({
        sourceId: r.id, url: r.url, eventTypes, description: r.description ?? undefined, disabledAtSource: !r.enabled, secret,
      });
      consumers.set(r.customer_id, consumer);
    }
    plan.consumers = [...consumers.values()];
    return plan;
  } finally {
    await db.end();
  }
}

// ---- 2. Check access and limits before writing anything ----
async function checkPlan(plan: Plan) {
  const res = await fetch(`${RELAY_API}/v1/outbound/access`, { headers: { Authorization: AUTH } });
  const access = await res.json();
  if (!access.enabled) throw new Error("The Outbound Gateway needs a Basic, Business or Pro plan: upgrade at https://my.webhookrelay.com/account.");
  const limits = access.limits as { consumers: number; endpoints_per_consumer: number; event_types: number };

  const haveConsumers = new Set((await relay.outbound.consumers.list()).map((c) => c.id));
  const haveTypes = new Set((await relay.outbound.eventTypes.list()).map((e) => e.name));
  const consumers = new Set([...haveConsumers, ...plan.consumers.map((c) => c.id)]).size;
  const types = new Set([...haveTypes, ...plan.eventTypes.map((e) => e.name)]).size;

  const problems: string[] = [];
  if (consumers > limits.consumers) problems.push(`${consumers} consumers exceed the limit of ${limits.consumers}`);
  if (types > limits.event_types) problems.push(`${types} event types exceed the limit of ${limits.event_types}`);
  for (const et of plan.eventTypes) if (!ID.test(et.name)) problems.push(`event type "${et.name}" is not a valid name`);
  for (const c of plan.consumers) {
    if (!ID.test(c.id)) problems.push(`consumer "${c.id}" is not a valid ID: map it to another customer ID`);
    if (c.endpoints.length > limits.endpoints_per_consumer)
      problems.push(`${c.id} has ${c.endpoints.length} endpoints, the limit is ${limits.endpoints_per_consumer}`);
    for (const e of c.endpoints) {
      if (!e.url.startsWith("https://")) problems.push(`${c.id} ${e.url}: endpoints must be public HTTPS`);
      if (e.eventTypes.length > 100) problems.push(`${c.id} ${e.url}: more than 100 event types`);
    }
  }
  for (const w of plan.warnings) console.warn(`warning: ${w}`);
  if (problems.length) {
    console.error(`Not importing:\n- ${problems.join("\n- ")}`);
    process.exit(1);
  }
}

// ---- 3. Import into Webhook Relay ----
const state: Record<string, string> = existsSync(STATE_FILE) ? JSON.parse(readFileSync(STATE_FILE, "utf8")) : {};

class HTTPError extends Error {
  constructor(readonly status: number, message: string) { super(message); }
}
const statusOf = (err: unknown) =>
  err instanceof WebhookRelayAPIError || err instanceof HTTPError ? err.status : undefined;

// REST create, so the existing secret can be passed (the SDKs add a `secret` option in their next release).
async function createEndpoint(consumer: string, e: PlanEndpoint): Promise<{ id: string }> {
  const res = await fetch(`${RELAY_API}/v1/outbound/consumers/${encodeURIComponent(consumer)}/endpoints`, {
    method: "POST",
    headers: { Authorization: AUTH, "Content-Type": "application/json" },
    body: JSON.stringify({
      url: e.url, event_types: e.eventTypes, description: e.description, headers: e.headers, rate: e.rate,
      ...(e.secret ? { secret: e.secret } : {}),
    }),
  });
  // Error bodies never echo the secret; the success body contains it, so read only the ID.
  if (!res.ok) throw new HTTPError(res.status, `POST endpoint: ${res.status} ${await res.text()}`);
  const { id } = (await res.json()) as { id: string };
  return { id };
}

async function step<T>(what: string, fn: () => Promise<T>): Promise<T> {
  for (let attempt = 1; ; attempt++) {
    try {
      return await fn();
    } catch (err) {
      if (statusOf(err) === 429 && attempt < 5) {
        await new Promise((r) => setTimeout(r, attempt * 2000));
        continue;
      }
      console.error(`Stopped at ${what}: ${err instanceof Error ? err.message : String(err)}`);
      console.error(`Progress is saved in ${STATE_FILE}. Fix the cause and run the script again.`);
      process.exit(1);
    }
  }
}

async function importPlan(plan: Plan) {
  for (const et of plan.eventTypes) {
    await step(`event type ${et.name}`, () =>
      relay.outbound.eventTypes.upsert(et.name, { description: et.description, deprecated: et.deprecated }));
  }
  for (const c of plan.consumers) {
    await step(`consumer ${c.id}`, () => relay.outbound.consumers.upsert(c.id, { name: c.name }));
    for (const e of c.endpoints) {
      if (state[e.sourceId]) continue; // imported on an earlier run
      const { id, created } = await step(`endpoint ${c.id} ${e.url}`, async () => {
        try {
          return { ...(await createEndpoint(c.id, e)), created: true };
        } catch (err) {
          if (statusOf(err) === 409) {
            const existing = (await relay.outbound.endpoints.list(c.id)).find((x) => x.url === e.url);
            if (existing) return { id: existing.id, created: false }; // already there: leave its state alone
          }
          throw err;
        }
      });
      if (created) await step(`pause ${id}`, () => relay.outbound.endpoints.pause(id));
      state[e.sourceId] = id;
      writeFileSync(STATE_FILE, JSON.stringify(state, null, 2));
      const secretNote = e.secret ? "secret imported" : "new secret";
      console.log(`${c.id}: ${e.url} -> ${id} ${created ? `(created, paused, ${secretNote})` : "(already existed, unchanged)"}${e.disabledAtSource ? " [disabled in your DB]" : ""}`);
    }
  }
}

const plan = await exportPlan();
await checkPlan(plan);
const endpoints = plan.consumers.reduce((n, c) => n + c.endpoints.length, 0);
console.log(`Plan: ${plan.eventTypes.length} event types, ${plan.consumers.length} consumers, ${endpoints} endpoints.`);
console.log(`Secrets: ${endpoints - plan.newSecrets.length} import, ${plan.newSecrets.length} get a new whsec_ secret.`);
if (plan.newSecrets.length) console.log(`New secret for endpoint rows: ${plan.newSecrets.join(", ")}`);
if (DRY_RUN) console.log("Dry run: nothing was written. Run again with DRY_RUN=false to import.");
else await importPlan(plan);