Migrate from Convoy

const { Convoy } = require("convoy.js");

const convoy = new Convoy({ uri: process.env.CONVOY_URL, api_key: process.env.CONVOY_API_KEY, project_id: process.env.CONVOY_PROJECT_ID });

await convoy.events.createFanOutEvent({
  owner_id: "customer_42",
  event_type: "invoice.paid",
  idempotency_key: "invoice-paid:inv_123",
  data: { invoice_id: "inv_123", amount: 4900 },
});

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

const relay = new WebhookRelay(); // reads RELAY_KEY and RELAY_SECRET

await relay.outbound.messages.publish(
  { consumer: "customer_42", eventType: "invoice.paid", payload: { invoice_id: "inv_123", amount: 4900 } },
  { idempotencyKey: "invoice-paid:inv_123" },
);

// Before
_, err := c.Events.FanoutEvent(ctx, &convoy.CreateFanoutEventRequest{
    EventType:      "invoice.paid",
    OwnerID:        "customer_42",
    IdempotencyKey: "invoice-paid:inv_123",
    Data:           []byte(`{"invoice_id":"inv_123","amount":4900}`),
})

// After
_, err := api.PublishOutboundMessage(&webhookrelay.OutboundMessagePublishOptions{
    Consumer:       "customer_42",
    EventType:      "invoice.paid",
    Payload:        map[string]interface{}{"invoice_id": "inv_123", "amount": 4900},
    IdempotencyKey: "invoice-paid:inv_123",
})

# Before
curl -X POST "$CONVOY_URL/v1/projects/$CONVOY_PROJECT_ID/events/fanout" \
  -H "Authorization: Bearer $CONVOY_API_KEY" -H 'Content-Type: application/json' \
  -d '{"owner_id":"customer_42","event_type":"invoice.paid","idempotency_key":"invoice-paid:inv_123","data":{"invoice_id":"inv_123","amount":4900}}'

# After
curl --user "$RELAY_KEY:$RELAY_SECRET" -X POST https://my.webhookrelay.com/v1/outbound/messages \
  -H 'Content-Type: application/json' -H 'Idempotency-Key: invoice-paid:inv_123' \
  -d '{"consumer":"customer_42","event_type":"invoice.paid","payload":{"invoice_id":"inv_123","amount":4900}}'

await relay.outbound.consumers.upsert("customer_42", { name: "Acme" });
const endpoint = await relay.outbound.endpoints.create("customer_42", {
  url: "https://acme.example/webhooks",
  eventTypes: ["invoice.paid"],
  autoDisable: true,
});
// endpoint.secret is the customer's new whsec_ secret. Store it like a password.

# CONVOY_SECRET holds the endpoint's current secret. Never echo it.
jq -n --arg url "https://acme.example/webhooks" --arg secret "$CONVOY_SECRET" \
  '{url: $url, event_types: ["invoice.paid"], secret: $secret}' |
curl -s --user "$RELAY_KEY:$RELAY_SECRET" -X POST \
  https://my.webhookrelay.com/v1/outbound/consumers/customer_42/endpoints \
  -H 'Content-Type: application/json' --data @- | jq '{id, state}'

import { verifyWebhook } from "./verify-webhook.js"; // Standard Webhooks check from the receiving guide
import { verifyConvoy } from "./verify-convoy.js";   // your receiver's existing Convoy check

export function verify(rawBody, headers) {
  if (headers["webhook-signature"]) {
    // The imported Convoy secret, or the new whsec_ secret if it couldn't be imported.
    return verifyWebhook(rawBody, headers, process.env.WEBHOOK_SECRET_RELAY);
  }
  if (headers["x-convoy-signature"]) {
    return verifyConvoy(rawBody, headers["x-convoy-signature"], process.env.WEBHOOK_SECRET_CONVOY);
  }
  return false;
}

// migrate-convoy.mts
// Env: CONVOY_URL (e.g. https://us.getconvoy.cloud/api), CONVOY_API_KEY, CONVOY_PROJECT_ID,
//      RELAY_KEY, RELAY_SECRET (token with "Outbound: manage"), DRY_RUN=false to write.
import { existsSync, readFileSync, writeFileSync } from "node:fs";
import { WebhookRelay, WebhookRelayAPIError } from "@webhookrelay/sdk";

const CONVOY = `${process.env.CONVOY_URL}/v1/projects/${process.env.CONVOY_PROJECT_ID}`;
const DRY_RUN = process.env.DRY_RUN !== "false";
const STATE_FILE = "migration-state.json"; // source endpoint ID -> Webhook Relay endpoint ID. Never secrets.
const ID = /^[a-zA-Z0-9][a-zA-Z0-9_.:-]{0,127}$/;
const RELAY_AUTH = `Basic ${Buffer.from(`${process.env.RELAY_KEY}:${process.env.RELAY_SECRET}`).toString("base64")}`;

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 the state file
};

// Webhook Relay imports 32-100 standard base64 characters (padded or not) with an optional whsec_ prefix.
// Anything else (Convoy's 25-character generated secrets, URL-safe base64, UUIDs) gets a new generated secret.
function importableSecret(secret: unknown): string | undefined {
  if (typeof secret !== "string") return undefined;
  const encoded = secret.replace(/^whsec_/, "");
  const bare = encoded.replace(/={1,2}$/, "");
  if (encoded.length > 100 || !/^[A-Za-z0-9+/]{32,100}$/.test(bare) || bare.length % 4 === 1) return undefined;
  if (bare !== encoded && encoded.length % 4 !== 0) return undefined; // bad padding
  return secret;
}

// The current Convoy secret is the live entry without an expiry (rolled secrets get expires_at).
function currentSecret(secrets: any[] | undefined): string | undefined {
  const live = (secrets ?? []).filter((s) => !s.deleted_at);
  return (live.find((s) => !s.expires_at) ?? live.at(-1))?.value;
}
type Plan = {
  eventTypes: { name: string; description?: string; deprecated?: boolean }[];
  consumers: { id: string; name?: string; endpoints: PlanEndpoint[] }[];
  warnings: string[];
};

// ---- 1. Export from Convoy (read-only) ----
async function convoy<T>(path: string, query: Record<string, string> = {}): Promise<T> {
  const qs = new URLSearchParams(query).toString();
  const res = await fetch(`${CONVOY}${path}${qs ? `?${qs}` : ""}`, {
    headers: { Authorization: `Bearer ${process.env.CONVOY_API_KEY}` },
  });
  if (!res.ok) throw new Error(`Convoy GET ${path}: ${res.status} ${await res.text()}`);
  return ((await res.json()) as { data: T }).data;
}

async function* convoyList<T>(path: string, query: Record<string, string> = {}): AsyncGenerator<T> {
  let cursor: string | undefined;
  for (;;) {
    const page = await convoy<{ content: T[]; pagination?: { has_next_page: boolean; next_page_cursor: string } }>(
      path, { ...query, perPage: "100", direction: "next", ...(cursor ? { next_page_cursor: cursor } : {}) });
    yield* page.content ?? [];
    if (!page.pagination?.has_next_page) return;
    cursor = page.pagination.next_page_cursor;
  }
}

async function exportPlan(): Promise<Plan> {
  const plan: Plan = { eventTypes: [], consumers: [], warnings: [] };
  const types = new Map<string, { name: string; description?: string; deprecated?: boolean }>();
  for (const et of await convoy<any[]>("/event-types")) {
    types.set(et.name, { name: et.name, description: et.description || undefined, deprecated: Boolean(et.deprecated_at) });
  }
  const consumers = new Map<string, Plan["consumers"][number]>();
  for await (const ep of convoyList<any>("/endpoints")) {
    const consumerId = ep.owner_id || ep.name || ep.uid;
    const secret = importableSecret(currentSecret(ep.secrets));
    if (!secret) plan.warnings.push(`${consumerId} endpoint ${ep.uid}: secret doesn't fit the import format; the endpoint gets a new one`);
    const eventTypes = new Set<string>();
    for await (const sub of convoyList<any>("/subscriptions", { endpointId: ep.uid })) {
      for (const t of sub.filter_config?.event_types ?? ["*"]) eventTypes.add(t);
      if (sub.filter_config?.filter && JSON.stringify(sub.filter_config.filter) !== "{}")
        plan.warnings.push(`${consumerId} ${ep.url}: subscription filters; recreate them as a Function`);
      if (sub.function) plan.warnings.push(`${consumerId} ${ep.url}: subscription function; recreate it as a Function`);
    }
    if (ep.authentication) plan.warnings.push(`${consumerId} ${ep.url}: endpoint authentication; re-enter it as a custom header by hand`);
    const list = eventTypes.has("*") || eventTypes.size === 0 ? ["*"] : [...eventTypes];
    for (const t of list) if (t !== "*" && !types.has(t)) types.set(t, { name: t });
    const consumer = consumers.get(consumerId) ?? { id: consumerId, name: ep.name, endpoints: [] as PlanEndpoint[] };
    consumer.endpoints.push({
      sourceId: ep.uid, url: ep.url, eventTypes: list, description: ep.description || undefined,
      disabledAtSource: ep.status !== "active", secret,
    });
    consumers.set(consumerId, consumer);
  }
  plan.eventTypes = [...types.values()];
  plan.consumers = [...consumers.values()];
  return plan;
}

// ---- 2. Check access and limits before writing anything ----
async function checkPlan(plan: Plan) {
  const res = await fetch("https://my.webhookrelay.com/v1/outbound/access", { headers: { Authorization: RELAY_AUTH } });
  const access = await res.json();
  if (!access.enabled) throw new Error("The Outbound Gateway isn't available on this account: upgrade to Basic or above.");
  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 your own 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 RelayError extends Error {
  constructor(readonly status: number, message: string) { super(message); }
}

// REST create, because the SDKs expose `secret` only from their next release. Webhook Relay's errors never echo the secret.
async function createEndpoint(consumer: string, e: PlanEndpoint): Promise<{ id: string }> {
  const res = await fetch(`https://my.webhookrelay.com/v1/outbound/consumers/${encodeURIComponent(consumer)}/endpoints`, {
    method: "POST",
    headers: { Authorization: RELAY_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 } : {}),
    }),
  });
  if (!res.ok) throw new RelayError(res.status, `create endpoint: ${res.status} ${await res.text()}`);
  const { id } = (await res.json()) as { id: string }; // the response also has the secret: don't log or store it
  return { id };
}

async function step<T>(what: string, fn: () => Promise<T>): Promise<T> {
  for (let attempt = 1; ; attempt++) {
    try {
      return await fn();
    } catch (err) {
      const status = err instanceof WebhookRelayAPIError || err instanceof RelayError ? err.status : 0;
      if (status === 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 (err instanceof RelayError && err.status === 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 = created ? (e.secret ? ", secret imported" : ", NEW secret") : "";
      console.log(`${c.id}: ${e.url} -> ${id} ${created ? "(created, paused" : "(already existed, unchanged"}${secretNote})${e.disabledAtSource ? " [not active at Convoy]" : ""}`);
    }
  }
}

const plan = await exportPlan();
await checkPlan(plan);
const endpoints = plan.consumers.flatMap((c) => c.endpoints);
const imported = endpoints.filter((e) => e.secret).length;
console.log(`Plan: ${plan.eventTypes.length} event types, ${plan.consumers.length} consumers, ${endpoints.length} endpoints (${imported} with an imported secret, ${endpoints.length - imported} with a new one).`);
if (DRY_RUN) console.log("Dry run: nothing was written. Run again with DRY_RUN=false to import.");
else await importPlan(plan);