Constant Contact to Virtuous CRM

import express from 'express';

const app = express();

app.post('/constant-contact/webhook', express.json(), async (req, res) => {
  // Verify signature per Constant Contact's documentation
  if (!verifyConstantContactSignature(req)) {
    return res.status(401).send('Invalid signature');
  }

  const events = Array.isArray(req.body) ? req.body : [req.body];
  for (const event of events) {
    await db.virtuous_sync_queue.insert({
      customer_id: resolveCustomerFromAccountId(event.account_id),
      source: 'constant_contact',
      source_event_type: event.event_type,
      source_record_id: event.contact_id,
      payload: event,
      status: 'pending',
    });
  }

  res.status(200).send('OK');
});

async function pollConstantContactChanges(customerId) {
  const ccToken = await loadCustomerConstantContactToken(customerId);
  const lastPolledAt = await db.constant_contact_sync_state.getLastPolledAt(customerId);

  // Query Constant Contact for contacts modified since the last poll
  const response = await fetch(
    `https://api.cc.email/v3/contacts?updated_after=${encodeURIComponent(lastPolledAt)}&limit=500`,
    {
      headers: { Authorization: `Bearer ${ccToken}` },
    }
  );

  if (!response.ok) {
    throw new Error(`Constant Contact poll failed: ${response.status}`);
  }
  const page = await response.json();

  for (const contact of page.contacts) {
    await db.virtuous_sync_queue.insert({
      customer_id: customerId,
      source: 'constant_contact',
      source_event_type: 'poll',
      source_record_id: contact.contact_id,
      payload: contact,
      status: 'pending',
    });
  }

  // Update sync state
  await db.constant_contact_sync_state.setLastPolledAt(customerId, new Date().toISOString());
}

function mapToContactTransaction(ccContact) {
  return {
    referenceSource: 'Constant Contact',
    referenceId: ccContact.contact_id,
    firstName: ccContact.first_name,
    lastName: ccContact.last_name,
    emailType: 'Home Email',
    email: ccContact.email_address?.address,
    phoneType: 'Mobile Phone',
    phone: ccContact.phone_number,
    address1: ccContact.street_addresses?.[0]?.street,
    city: ccContact.street_addresses?.[0]?.city,
    state: ccContact.street_addresses?.[0]?.state,
    postal: ccContact.street_addresses?.[0]?.postal_code,
    country: ccContact.street_addresses?.[0]?.country ?? 'US',
    contactType: 'Household',
    tags: ccContact.list_memberships
      ?.map((lm) => `CC: ${listNameFromId(lm.list_id)}`)
      .join(', '),
    originSegmentCode: 'CONSTANT-CONTACT-SYNC',
  };
}

async function handleVirtuousContactUpdate(event) {
  const contact = event.data;

  const ccRef = (contact.contactReferences || []).find(
    (r) => r.source === 'Constant Contact'
  );
  if (!ccRef) return;

  const currentTags = new Set(contact.tags?.map((t) => t.tag) || []);
  const previousTags = await db.contact_sync_state.getLastKnownTags(contact.id);

  // For each tag prefixed 'CC:' that was added, add the contact to that list
  const added = [...currentTags].filter((t) => !previousTags.has(t) && t.startsWith('CC:'));
  const removed = [...previousTags].filter((t) => !currentTags.has(t) && t.startsWith('CC:'));

  for (const tag of added) {
    const listId = listIdFromTagName(tag);
    await constantContactClient.addToList(ccRef.id, listId);
  }
  for (const tag of removed) {
    const listId = listIdFromTagName(tag);
    await constantContactClient.removeFromList(ccRef.id, listId);
  }

  await db.contact_sync_state.setTags(contact.id, [...currentTags]);
}

async function bulkImportConstantContactContacts(customerId) {
  const ccToken = await loadCustomerConstantContactToken(customerId);
  let cursor = null;

  do {
    const url = cursor
      ? `https://api.cc.email/v3/contacts?cursor=${encodeURIComponent(cursor)}&limit=500`
      : `https://api.cc.email/v3/contacts?limit=500`;

    const response = await fetch(url, { headers: { Authorization: `Bearer ${ccToken}` } });
    const page = await response.json();

    for (const contact of page.contacts) {
      await db.virtuous_sync_queue.insert({
        customer_id: customerId,
        source: 'constant_contact',
        source_event_type: 'bulk_import',
        source_record_id: contact.contact_id,
        payload: contact,
        status: 'pending',
      });
    }

    cursor = page._links?.next?.href;
  } while (cursor);
}