Salesforce Event-Driven Integration: Architecture Guide and Implementation

// Create PushTopic in Salesforce (via Developer Console or Apex)
PushTopic pushTopic = new PushTopic();
pushTopic.Name = 'AccountUpdates';
pushTopic.Query = 'SELECT Id, Name, Industry, AnnualRevenue FROM Account WHERE Industry != null';
pushTopic.ApiVersion = 65.0;
pushTopic.NotifyForOperationCreate = true;
pushTopic.NotifyForOperationUpdate = true;
pushTopic.NotifyForOperationDelete = false;
pushTopic.NotifyForOperationUndelete = false;
pushTopic.NotifyForFields = 'Referenced'; // Only notify if queried fields change

insert pushTopic;

const faye = require('faye');
const jsforce = require('jsforce');

// Authenticate with Salesforce
const conn = new jsforce.Connection({
  loginUrl: 'https://login.salesforce.com'
});

await conn.login(username, password);

// Create CometD client
const client = new faye.Client(
  `${conn.instanceUrl}/cometd/65.0/`,
  { timeout: 120 }
);

// Add authentication extension
client.addExtension({
  outgoing: (message, callback) => {
    message.ext = message.ext || {};
    message.ext.replay = true; // Enable event replay

    if (message.channel === '/meta/handshake') {
      message.ext.authorization = `Bearer ${conn.accessToken}`;
    }

    callback(message);
  }
});

// Subscribe to PushTopic
const subscription = client.subscribe(
  '/topic/AccountUpdates',
  (message) => {
    console.log('Received event:', message);

    // Event payload structure
    const event = message.data.event;
    const sobject = message.data.sobject;

    // React to the change
    handleAccountUpdate(sobject);
  }
);

// Handle errors
subscription.then(
  () => console.log('Subscribed to AccountUpdates'),
  (error) => console.error('Subscription failed:', error)
);

// Process the received data
const handleAccountUpdate = (account) => {
  console.log(`Account ${account.Id} updated:`);
  console.log(`  Name: ${account.Name}`);
  console.log(`  Industry: ${account.Industry}`);
  console.log(`  Revenue: ${account.AnnualRevenue}`);

  // Update external system
  syncToExternalSystem(account);
};

// Store replay ID after processing each event
let lastReplayId = -1; // -1 means get all available events

client.addExtension({
  outgoing: (message, callback) => {
    if (message.channel === '/meta/subscribe') {
      message.ext = message.ext || {};
      message.ext.replay = {
        '/topic/AccountUpdates': lastReplayId
      };
    }
    callback(message);
  }
});

// Update replay ID when receiving events
client.subscribe('/topic/AccountUpdates', (message) => {
  lastReplayId = message.event.replayId;

  // Store this ID persistently so recovery works after restarts
  saveReplayId(lastReplayId);

  handleAccountUpdate(message.data.sobject);
});

// Publish event when order ships
public static void notifyOrderShipped(String orderId) {
    Order order = [SELECT OrderNumber, Customer__r.Email, Tracking_Number__c,
                   Shipping_Method__c, Estimated_Delivery__c
                   FROM Order WHERE Id = :orderId];

    // Create event
    Order_Shipped__e event = new Order_Shipped__e(
        Order_Number__c = order.OrderNumber,
        Customer_Email__c = order.Customer__r.Email,
        Tracking_Number__c = order.Tracking_Number__c,
        Shipping_Method__c = order.Shipping_Method__c,
        Estimated_Delivery__c = order.Estimated_Delivery__c
    );

    // Publish event
    EventBus.publish(event);
}

// External system publishes event to Salesforce
const publishEvent = async (eventData) => {
  const response = await fetch(
    `${instanceUrl}/services/data/v65.0/sobjects/Order_Shipped__e/`,
    {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${accessToken}`,
        'Content-Type': 'application/json'
      },
      body: JSON.stringify(eventData)
    }
  );

  return await response.json();
};

// Usage
await publishEvent({
  Order_Number__c: 'ORD-12345',
  Customer_Email__c: '[email protected]',
  Tracking_Number__c: '1Z999AA10123456784',
  Shipping_Method__c: 'Express',
  Estimated_Delivery__c: '2025-06-15'
});

trigger OrderShippedSubscriber on Order_Shipped__e (after insert) {
    List<Task> followUpTasks = new List<Task>();

    for (Order_Shipped__e event : Trigger.New) {
        // Create follow-up task for account manager
        Task task = new Task(
            Subject = 'Order ' + event.Order_Number__c + ' shipped',
            Description = 'Order shipped via ' + event.Shipping_Method__c +
                          '. Tracking: ' + event.Tracking_Number__c,
            Priority = 'Normal',
            Status = 'Not Started',
            ActivityDate = event.Estimated_Delivery__c
        );

        followUpTasks.add(task);
    }

    if (!followUpTasks.isEmpty()) {
        insert followUpTasks;
    }
}

// Subscribe to Platform Events from external system
client.subscribe('/event/Order_Shipped__e', (message) => {
  const event = message.data.payload;

  console.log(`Order ${event.Order_Number__c} shipped`);

  // Send customer notification email
  sendShippingNotificationEmail({
    to: event.Customer_Email__c,
    orderNumber: event.Order_Number__c,
    trackingNumber: event.Tracking_Number__c,
    estimatedDelivery: event.Estimated_Delivery__c
  });

  // Update external order tracking system
  updateOrderTrackingSystem({
    orderNumber: event.Order_Number__c,
    status: 'SHIPPED',
    trackingNumber: event.Tracking_Number__c
  });
});

// Subscribe to Account change events
client.subscribe('/data/AccountChangeEvent', (message) => {
  const changeEvent = message.data.payload;
  const header = changeEvent.ChangeEventHeader;

  console.log('Change Type:', header.changeType); // CREATE, UPDATE, DELETE, UNDELETE
  console.log('Changed Fields:', header.changedFields);
  console.log('Record IDs:', header.recordIds);

  // Process based on change type
  switch (header.changeType) {
    case 'CREATE':
      handleAccountCreated(changeEvent);
      break;
    case 'UPDATE':
      handleAccountUpdated(changeEvent, header.changedFields);
      break;
    case 'DELETE':
      handleAccountDeleted(header.recordIds);
      break;
  }
});

const handleAccountUpdated = (event, changedFields) => {
  // CDC provides both old and new values for changed fields
  console.log('Account ID:', event.Id);

  changedFields.forEach(field => {
    console.log(`${field} changed:`);
    console.log(`  Old value: ${event[field]}`);
    // New value is in the main event payload
  });

  // Sync to data warehouse
  syncToDataWarehouse({
    table: 'accounts',
    operation: 'UPDATE',
    recordId: event.Id,
    changes: changedFields.reduce((acc, field) => {
      acc[field] = event[field];
      return acc;
    }, {})
  });
};