Integrate with ActiveCampaign

const { AC_API_URL, AC_API_KEY, AC_LIST_ID } = process.env;

// Queue every request so each one starts at least 200 ms after the last, even
// when several webhooks arrive at once. This stays under 5 requests per second.
let queue = Promise.resolve();

function nextSlot() {
  const slot = queue;
  queue = queue.then(() => new Promise((resolve) => setTimeout(resolve, 200)));
  return slot;
}

async function ac(path, { method = "GET", body } = {}) {
  await nextSlot();

  const res = await fetch(`${AC_API_URL}/api/3${path}`, {
    method,
    headers: { "Api-Token": AC_API_KEY, "Content-Type": "application/json" },
    body: body && JSON.stringify(body),
  });
  if (!res.ok) throw new Error(`ActiveCampaign ${method} ${path}: ${res.status}`);
  const text = await res.text();
  return text ? JSON.parse(text) : null;
}

// Mighty leaves out an email the member hasn't agreed to share, and masks it
// (a***@***.***) when your plan or token can't see it. Skip both.
export function usableEmail(email) {
  return email && !email.includes("*") ? email : null;
}

// Create the contact, or update the one with this email. Returns its ID.
export async function syncContact({ email, firstName, lastName }) {
  const { contact } = await ac("/contact/sync", {
    method: "POST",
    body: { contact: { email, firstName, lastName } },
  });
  return contact.id;
}

export async function subscribeToList(contactId) {
  await ac("/contactLists", {
    method: "POST",
    body: { contactList: { list: AC_LIST_ID, contact: contactId, status: 1 } },
  });
}

const tagIds = new Map();

async function tagId(name, { create = true } = {}) {
  if (tagIds.has(name)) return tagIds.get(name);
  const { tags } = await ac(`/tags?search=${encodeURIComponent(name)}`);
  let tag = tags.find((t) => t.tag === name);
  if (!tag) {
    if (!create) return null;
    ({ tag } = await ac("/tags", {
      method: "POST",
      body: { tag: { tag: name, tagType: "contact", description: "Synced from Mighty Networks" } },
    }));
  }
  tagIds.set(name, tag.id);
  return tag.id;
}

export async function addTag(contactId, name) {
  await ac("/contactTags", {
    method: "POST",
    body: { contactTag: { contact: contactId, tag: await tagId(name) } },
  });
}

export async function removeTag(contactId, name) {
  const id = await tagId(name, { create: false });
  if (!id) return;
  const { contactTags } = await ac(`/contacts/${contactId}/contactTags`);
  const match = contactTags.find((ct) => String(ct.tag) === String(id));
  if (match) await ac(`/contactTags/${match.id}`, { method: "DELETE" });
}

import crypto from "node:crypto";
import express from "express";
import { usableEmail, syncContact, subscribeToList, addTag, removeTag } from "./activecampaign.js";

const { WEBHOOK_SECRET } = process.env;
// Without a secret, the expected header would be "Bearer undefined", which anyone can send.
if (!WEBHOOK_SECRET) throw new Error("Set WEBHOOK_SECRET before starting the server");
const app = express();

// Find or create the ActiveCampaign contact for a member in a webhook payload.
async function contactFor(member) {
  const email = usableEmail(member.email);
  if (!email) return null;
  return syncContact({ email, firstName: member.first_name, lastName: member.last_name });
}

const handlers = {
  // Fires for the Network and for each Space. A Network join has space_id equal to network_id.
  async MemberJoinedHook({ member, space_id, network_id }) {
    const contactId = await contactFor(member);
    if (!contactId) return;
    if (space_id === network_id) {
      await subscribeToList(contactId);
      await removeTag(contactId, "Mighty: Former member");
      await addTag(contactId, "Mighty: Member");
    } else {
      await addTag(contactId, `Mighty Space: ${space_id}`);
    }
  },

  async MemberLeftHook({ member, space_id, network_id }) {
    const contactId = await contactFor(member);
    if (!contactId) return;
    if (space_id === network_id) {
      await removeTag(contactId, "Mighty: Member");
      await addTag(contactId, "Mighty: Former member");
    } else {
      await removeTag(contactId, `Mighty Space: ${space_id}`);
    }
  },

  async MemberUpdatedHook({ member }) {
    await contactFor(member);
  },

  async MemberTagAddedHook({ member, tag }) {
    const contactId = await contactFor(member);
    if (contactId) await addTag(contactId, `Mighty tag: ${tag.title}`);
  },

  async MemberTagRemovedHook({ member, tag }) {
    const contactId = await contactFor(member);
    if (contactId) await removeTag(contactId, `Mighty tag: ${tag.title}`);
  },

  // Purchase and cancellation payloads carry member fields at the top level.
  async MemberPurchasedHook(p) {
    const email = usableEmail(p.member_email);
    if (!email) return;
    const contactId = await syncContact({ email, firstName: p.member_first_name, lastName: p.member_last_name });
    // plan.name is optional; fall back to plan.id so the tag is never "Mighty plan: undefined".
    await addTag(contactId, `Mighty plan: ${p.plan.name ?? p.plan.id}`);
  },

  async MemberSubscriptionCanceledHook(p) {
    const email = usableEmail(p.email);
    if (!email) return;
    const contactId = await syncContact({ email, firstName: p.first_name, lastName: p.last_name });
    const plan = p.plan.name ?? p.plan.id;
    await removeTag(contactId, `Mighty plan: ${plan}`);
    await addTag(contactId, `Mighty canceled: ${plan}`);
  },

  async MemberCourseProgressCompletedHook({ member, course }) {
    const contactId = await contactFor(member);
    if (contactId) await addTag(contactId, `Mighty course completed: ${course.name}`);
  },
};

// Deliveries can repeat. Replace this with a durable store (a database table, Redis) in production.
const processed = new Set();

app.post("/webhooks/mighty", express.json(), async (req, res) => {
  const expected = Buffer.from(`Bearer ${WEBHOOK_SECRET}`);
  const received = Buffer.from(req.get("authorization") ?? "");
  if (received.length !== expected.length || !crypto.timingSafeEqual(received, expected)) {
    return res.sendStatus(401);
  }

  const { event_id, event_type, payload } = req.body;
  const handler = handlers[event_type];
  if (!handler || processed.has(event_id)) return res.sendStatus(200);

  try {
    await handler(payload);
    processed.add(event_id);
    res.sendStatus(200);
  } catch (err) {
    console.error(`Failed to process ${event_type} ${event_id}`, err);
    res.sendStatus(500); // Any non-2xx response makes Mighty retry the delivery later.
  }
});

app.listen(process.env.PORT ?? 3000);

import { readFile, writeFile } from "node:fs/promises";

const { MIGHTY_NETWORK, MIGHTY_CLIENT_ID, MIGHTY_CLIENT_SECRET, MIGHTY_REFRESH_TOKEN } = process.env;
const TOKEN_FILE = new URL("./.mighty-token.json", import.meta.url);
const USER_AGENT = "activecampaign-sync/1.0 (+https://your-server.example.com)";

let cached = null; // { accessToken, expiresAt }
let refreshing = null;

async function currentRefreshToken() {
  try {
    return JSON.parse(await readFile(TOKEN_FILE, "utf8")).refresh_token;
  } catch {
    return MIGHTY_REFRESH_TOKEN; // First run: fall back to the token from Step 2.
  }
}

async function refresh() {
  const res = await fetch(`https://${MIGHTY_NETWORK}.mn.co/oauth/token`, {
    method: "POST",
    headers: { "Content-Type": "application/x-www-form-urlencoded", "User-Agent": USER_AGENT },
    body: new URLSearchParams({
      grant_type: "refresh_token",
      refresh_token: await currentRefreshToken(),
      client_id: MIGHTY_CLIENT_ID,
      client_secret: MIGHTY_CLIENT_SECRET,
    }),
  });
  if (!res.ok) throw new Error(`Token refresh failed: ${res.status}`);
  const token = await res.json();
  await writeFile(TOKEN_FILE, JSON.stringify({ refresh_token: token.refresh_token }));
  cached = { accessToken: token.access_token, expiresAt: Date.now() + (token.expires_in - 60) * 1000 };
  return cached.accessToken;
}

export async function getAccessToken() {
  if (cached && Date.now() < cached.expiresAt) return cached.accessToken;
  // Share one in-flight refresh across concurrent callers instead of racing.
  if (!refreshing) refreshing = refresh().finally(() => (refreshing = null));
  return refreshing;
}

export async function mighty(query, variables) {
  const res = await fetch(`https://api.mn.co/networks/${MIGHTY_NETWORK}/graphql`, {
    method: "POST",
    headers: {
      Authorization: `Bearer ${await getAccessToken()}`,
      "Content-Type": "application/json",
      "User-Agent": USER_AGENT,
    },
    body: JSON.stringify({ query, variables }),
  });
  // Bot protection (403) and rate limits (429) return non-JSON bodies.
  if (!res.ok) throw new Error(`Mighty API request failed: ${res.status}`);
  const { data, errors } = await res.json();
  if (errors) throw new Error(JSON.stringify(errors));
  return data;
}

import { usableEmail, syncContact, subscribeToList, addTag } from "./activecampaign.js";
import { mighty } from "./mighty.js";

const MEMBERS_QUERY = `
  query Members($after: String) {
    network {
      members(first: 50, after: $after, sort: RESOURCE_ID) {
        pageInfo { hasNextPage endCursor }
        nodes { resourceId email firstName lastName tags { title } }
      }
    }
  }
`;

let after = null;
let synced = 0;

do {
  const { network } = await mighty(MEMBERS_QUERY, { after });
  const { nodes, pageInfo } = network.members;
  for (const member of nodes) {
    const email = usableEmail(member.email);
    if (!email) continue;
    const contactId = await syncContact({ email, firstName: member.firstName, lastName: member.lastName });
    await subscribeToList(contactId);
    await addTag(contactId, "Mighty: Member");
    for (const tag of member.tags) await addTag(contactId, `Mighty tag: ${tag.title}`);
    synced += 1;
  }

  after = pageInfo.hasNextPage ? pageInfo.endCursor : null;
} while (after);

console.log(`Synced ${synced} members to ActiveCampaign`);