Event-Driven Integration Patterns

Event-Driven Integration Patterns

This document provides detailed implementation patterns for Platform Events and Change Data Capture (CDC) in Salesforce integrations.

Parent Document: sf-integration/SKILL.md Related: callout-patterns.md

Table of Contents

• Platform Events
  • Platform Event Definition
  • Event Publisher
  • Event Subscriber Trigger
  • High-Volume vs Standard-Volume Events
• Change Data Capture (CDC)
  • CDC Enablement
  • CDC Subscriber Trigger
  • CDC Handler Service
  • Field-Specific Change Detection

Platform Events

Platform Events enable asynchronous, event-driven communication between applications. They provide a publish-subscribe model where publishers fire events and subscribers listen for them.

Platform Event Definition

Use Case: Asynchronous, event-driven communication

Template: templates/platform-events/platform-event-definition.object-meta.xml

Standard Volume Event

Best for moderate event volumes (~2,000 events/hour):

<?xml version="1.0" encoding="UTF-8"?>
<CustomObject xmlns="http://soap.sforce.com/2006/04/metadata">
    <deploymentStatus>Deployed</deploymentStatus>
    <eventType>StandardVolume</eventType>
    <label>{{EventLabel}}</label>
    <pluralLabel>{{EventPluralLabel}}</pluralLabel>
    <publishBehavior>PublishAfterCommit</publishBehavior>

    <fields>
        <fullName>{{FieldName}}__c</fullName>
        <label>{{FieldLabel}}</label>
        <type>Text</type>
        <length>255</length>
    </fields>

    <fields>
        <fullName>RecordId__c</fullName>
        <label>Record ID</label>
        <type>Text</type>
        <length>18</length>
        <description>Salesforce record ID related to this event</description>
    </fields>

    <fields>
        <fullName>Timestamp__c</fullName>
        <label>Timestamp</label>
        <type>DateTime</type>
        <description>When the event was triggered</description>
    </fields>
</CustomObject>

High-Volume Event

Best for millions of events per day:

<?xml version="1.0" encoding="UTF-8"?>
<CustomObject xmlns="http://soap.sforce.com/2006/04/metadata">
    <deploymentStatus>Deployed</deploymentStatus>
    <eventType>HighVolume</eventType>
    <label>{{EventLabel}}</label>
    <pluralLabel>{{EventPluralLabel}}</pluralLabel>
    <publishBehavior>PublishAfterCommit</publishBehavior>

    <fields>
        <fullName>{{FieldName}}__c</fullName>
        <label>{{FieldLabel}}</label>
        <type>Text</type>
        <length>255</length>
    </fields>
</CustomObject>

Key Configuration Options

PublishBehavior Details:

• PublishAfterCommit: Event published only if transaction commits (recommended)
• PublishImmediately: Event published immediately, even if transaction rolls back

Event Publisher

Template: templates/platform-events/event-publisher.cls

Bulk Event Publisher

public with sharing class {{EventName}}Publisher {

    public static void publishEvents(List<{{EventName}}__e> events) {
        if (events == null || events.isEmpty()) {
            return;
        }

        List<Database.SaveResult> results = EventBus.publish(events);

        for (Integer i = 0; i < results.size(); i++) {
            Database.SaveResult sr = results[i];
            if (!sr.isSuccess()) {
                for (Database.Error err : sr.getErrors()) {
                    System.debug(LoggingLevel.ERROR,
                        'Event publish error: ' + err.getStatusCode() + ' - ' + err.getMessage());
                }
            }
        }
    }

    public static void publishSingleEvent(Map<String, Object> eventData) {
        {{EventName}}__e event = new {{EventName}}__e();

        // Map fields from eventData
        event.{{FieldName}}__c = (String) eventData.get('{{fieldKey}}');
        event.RecordId__c = (String) eventData.get('recordId');
        event.Timestamp__c = DateTime.now();

        Database.SaveResult sr = EventBus.publish(event);
        if (!sr.isSuccess()) {
            throw new EventPublishException('Failed to publish event: ' + sr.getErrors());
        }
    }

    public class EventPublishException extends Exception {}
}

Usage Examples

Single Event:

Map<String, Object> eventData = new Map<String, Object>{
    'recordId' => '001xx000003DXXXAAA',
    'status' => 'Completed',
    'amount' => 1000.00
};
OrderStatusPublisher.publishSingleEvent(eventData);

Bulk Events:

List<Order_Status__e> events = new List<Order_Status__e>();

for (Order order : orders) {
    Order_Status__e event = new Order_Status__e();
    event.RecordId__c = order.Id;
    event.Status__c = order.Status;
    event.Timestamp__c = DateTime.now();
    events.add(event);
}

OrderStatusPublisher.publishEvents(events);

Best Practices for Publishing

1. Batch Events: Publish up to 2,000 events per transaction (governor limit)
2. Error Handling: Always check Database.SaveResult for failures
3. Transaction Context: Use PublishAfterCommit to ensure events only fire on successful transactions
4. Field Population: Populate all required fields before publishing
5. Logging: Log failed event publishes for debugging

Event Subscriber Trigger

Template: templates/platform-events/event-subscriber-trigger.trigger

Standard Volume Subscriber

trigger {{EventName}}Subscriber on {{EventName}}__e (after insert) {
    // Get replay ID for resumption
    String lastReplayId = '';

    for ({{EventName}}__e event : Trigger.new) {
        // Store replay ID for potential resume
        lastReplayId = event.ReplayId;

        try {
            // Process event
            {{EventName}}Handler.processEvent(event);
        } catch (Exception e) {
            // Log error but don't throw - allow other events to process
            System.debug(LoggingLevel.ERROR,
                'Event processing error: ' + e.getMessage() +
                ' ReplayId: ' + event.ReplayId);
        }
    }
}

High-Volume Subscriber (with Resume Checkpoint)

trigger {{EventName}}Subscriber on {{EventName}}__e (after insert) {
    String lastReplayId = '';

    for ({{EventName}}__e event : Trigger.new) {
        lastReplayId = event.ReplayId;

        try {
            {{EventName}}Handler.processEvent(event);
        } catch (Exception e) {
            System.debug(LoggingLevel.ERROR,
                'Event processing error: ' + e.getMessage() +
                ' ReplayId: ' + event.ReplayId);
        }
    }

    // Set resume checkpoint for high-volume events
    // Allows resuming from this point if subscriber fails
    EventBus.TriggerContext.currentContext().setResumeCheckpoint(lastReplayId);
}

Event Handler Class

public with sharing class {{EventName}}Handler {

    public static void processEvent({{EventName}}__e event) {
        // Extract event data
        String recordId = event.RecordId__c;
        String status = event.Status__c;
        DateTime timestamp = event.Timestamp__c;

        System.debug('Processing event - RecordId: ' + recordId +
                     ', Status: ' + status +
                     ', ReplayId: ' + event.ReplayId);

        // Business logic
        updateRelatedRecords(recordId, status);
        syncToExternalSystem(recordId, status);
    }

    private static void updateRelatedRecords(String recordId, String status) {
        // Update related records based on event
        List<Task> tasks = [
            SELECT Id, Status
            FROM Task
            WHERE WhatId = :recordId
            WITH USER_MODE
        ];

        for (Task t : tasks) {
            t.Status = status;
        }

        update as user tasks;
    }

    private static void syncToExternalSystem(String recordId, String status) {
        // Queue async callout
        Map<String, Object> payload = new Map<String, Object>{
            'recordId' => recordId,
            'status' => status
        };
        System.enqueueJob(new ExternalSyncQueueable(payload));
    }
}

High-Volume vs Standard-Volume Events

When to Use High-Volume Events

• IoT Data: Sensor data, device telemetry
• Real-Time Analytics: Clickstream, user behavior tracking
• High-Traffic Systems: E-commerce order processing, stock updates
• Event Sourcing: Append-only event logs

When to Use Standard-Volume Events

• Business Workflows: Order status updates, approval processes
• Integration Events: Sync to external CRM, ERP systems
• Notifications: Email triggers, Slack notifications
• Audit Trails: Record-level change notifications

Change Data Capture (CDC)

Change Data Capture publishes change events whenever records are created, updated, deleted, or undeleted in Salesforce. CDC events are published automatically—no custom code needed.

CDC Enablement

Enable via Setup UI

1. Navigate to Setup → Integrations → Change Data Capture
2. Select objects to enable (Standard or Custom)
3. Save

Enable via Metadata API

File: force-app/main/default/changeDataCaptures/AccountChangeEvent.cdc-meta.xml

<?xml version="1.0" encoding="UTF-8"?>
<ChangeDataCapture xmlns="http://soap.sforce.com/2006/04/metadata">
    <entityName>Account</entityName>
    <isEnabled>true</isEnabled>
</ChangeDataCapture>

Channel Naming Convention

CDC channels follow the pattern: {{ObjectAPIName}}ChangeEvent

Examples:

• AccountChangeEvent (Standard object)
• Order__ChangeEvent (Custom object)
• OpportunityChangeEvent
• Contact_Request__ChangeEvent

CDC Subscriber Trigger

Template: templates/cdc/cdc-subscriber-trigger.trigger

Basic CDC Subscriber

trigger {{ObjectName}}CDCSubscriber on {{ObjectName}}ChangeEvent (after insert) {

    for ({{ObjectName}}ChangeEvent event : Trigger.new) {
        // Get change event header
        EventBus.ChangeEventHeader header = event.ChangeEventHeader;

        String changeType = header.getChangeType();
        List<String> changedFields = header.getChangedFields();
        String recordId = header.getRecordIds()[0]; // First record ID

        System.debug('CDC Event - Type: ' + changeType +
                     ', RecordId: ' + recordId +
                     ', Changed Fields: ' + changedFields);

        // Route based on change type
        switch on changeType {
            when 'CREATE' {
                // Handle new record
                {{ObjectName}}CDCHandler.handleCreate(event);
            }
            when 'UPDATE' {
                // Handle update
                {{ObjectName}}CDCHandler.handleUpdate(event, changedFields);
            }
            when 'DELETE' {
                // Handle delete
                {{ObjectName}}CDCHandler.handleDelete(recordId);
            }
            when 'UNDELETE' {
                // Handle undelete
                {{ObjectName}}CDCHandler.handleUndelete(event);
            }
        }
    }
}

ChangeEventHeader Fields

CDC Handler Service

Template: templates/cdc/cdc-handler.cls

public with sharing class {{ObjectName}}CDCHandler {

    public static void handleCreate({{ObjectName}}ChangeEvent event) {
        // Sync to external system on create
        Map<String, Object> payload = buildPayload(event);
        System.enqueueJob(new ExternalSystemSyncQueueable(payload, 'CREATE'));
    }

    public static void handleUpdate({{ObjectName}}ChangeEvent event, List<String> changedFields) {
        // Only sync if relevant fields changed
        Set<String> fieldsToWatch = new Set<String>{'Name', 'Status__c', 'Amount__c'};

        Boolean relevantChange = false;
        for (String field : changedFields) {
            if (fieldsToWatch.contains(field)) {
                relevantChange = true;
                break;
            }
        }

        if (relevantChange) {
            Map<String, Object> payload = buildPayload(event);
            payload.put('changedFields', changedFields);
            System.enqueueJob(new ExternalSystemSyncQueueable(payload, 'UPDATE'));
        }
    }

    public static void handleDelete(String recordId) {
        Map<String, Object> payload = new Map<String, Object>{'recordId' => recordId};
        System.enqueueJob(new ExternalSystemSyncQueueable(payload, 'DELETE'));
    }

    public static void handleUndelete({{ObjectName}}ChangeEvent event) {
        handleCreate(event); // Treat undelete like create
    }

    private static Map<String, Object> buildPayload({{ObjectName}}ChangeEvent event) {
        return new Map<String, Object>{
            'recordId' => event.ChangeEventHeader.getRecordIds()[0],
            'commitTimestamp' => event.ChangeEventHeader.getCommitTimestamp(),
            'commitUser' => event.ChangeEventHeader.getCommitUser(),
            // Add event field values
            'name' => event.Name,
            'status' => event.Status__c
            // Add more fields
        };
    }
}

Field-Specific Change Detection

Filtering by Changed Fields

public static void handleUpdate(AccountChangeEvent event, List<String> changedFields) {
    // Only process if billing address changed
    Set<String> billingFields = new Set<String>{
        'BillingStreet',
        'BillingCity',
        'BillingState',
        'BillingPostalCode',
        'BillingCountry'
    };

    Boolean billingChanged = false;
    for (String field : changedFields) {
        if (billingFields.contains(field)) {
            billingChanged = true;
            break;
        }
    }

    if (billingChanged) {
        updateShippingPartner(event);
    }
}

Multi-Field Change Logic

public static void handleUpdate(OpportunityChangeEvent event, List<String> changedFields) {
    Set<String> changedFieldSet = new Set<String>(changedFields);

    // Check if stage AND amount both changed
    if (changedFieldSet.contains('StageName') && changedFieldSet.contains('Amount')) {
        // Alert sales ops about significant deal change
        sendAlert('Deal stage and amount changed', event);
    }

    // Check if close date moved backward
    if (changedFieldSet.contains('CloseDate')) {
        checkCloseDateRegression(event);
    }
}

CDC vs Platform Events: When to Use Which

Decision Matrix

┌───────────────────────────────────────────────────────────────────────┐
│  WHEN TO USE PLATFORM EVENTS vs CHANGE DATA CAPTURE                   │
├───────────────────────────────────────────────────────────────────────┤
│  Use PLATFORM EVENTS when:                                            │
│  • Custom business event (not tied to record changes)                 │
│  • Event needs custom fields not on object                            │
│  • Need to batch/aggregate data before publishing                     │
│  • Publishing from external system to Salesforce                      │
│  • Complex event logic (multi-object aggregation)                     │
│                                                                        │
│  Use CHANGE DATA CAPTURE when:                                        │
│  • Syncing record changes to external system                          │
│  • Audit trail of all record modifications                            │
│  • Real-time replication to data warehouse                            │
│  • Event sourcing from Salesforce objects                             │
│  • Zero-code event publishing required                                │
└───────────────────────────────────────────────────────────────────────┘

Advanced Patterns

Combining CDC with Callouts

public with sharing class AccountCDCHandler {

    public static void handleUpdate(AccountChangeEvent event, List<String> changedFields) {
        // Extract data
        String recordId = event.ChangeEventHeader.getRecordIds()[0];
        String accountName = event.Name;

        // Queue async callout to external CRM
        Map<String, Object> payload = new Map<String, Object>{
            'salesforceId' => recordId,
            'name' => accountName,
            'changedFields' => changedFields,
            'timestamp' => event.ChangeEventHeader.getCommitTimestamp()
        };

        System.enqueueJob(new CRMSyncQueueable(payload));
    }
}

Event Replay with Stored ReplayId

public with sharing class EventReplayService {

    @future(callout=true)
    public static void replayFromLastCheckpoint(String eventChannel) {
        // Get last stored replay ID
        Event_Checkpoint__c checkpoint = [
            SELECT ReplayId__c
            FROM Event_Checkpoint__c
            WHERE Channel__c = :eventChannel
            LIMIT 1
        ];

        if (checkpoint != null) {
            // Use stored replay ID to resume from last successful event
            // Note: Replay is typically done via API/streaming API, not Apex
            System.debug('Last replay ID: ' + checkpoint.ReplayId__c);
        }
    }

    public static void storeCheckpoint(String channel, String replayId) {
        Event_Checkpoint__c checkpoint = new Event_Checkpoint__c(
            Channel__c = channel,
            ReplayId__c = replayId,
            Last_Updated__c = DateTime.now()
        );
        upsert checkpoint Channel__c;
    }
}

Error Handling with Dead Letter Queue

trigger OrderEventSubscriber on Order_Status__e (after insert) {
    for (Order_Status__e event : Trigger.new) {
        try {
            OrderEventHandler.processEvent(event);
        } catch (Exception e) {
            // Log to dead letter queue for manual review
            Event_Error_Log__c errorLog = new Event_Error_Log__c(
                Event_Type__c = 'Order_Status__e',
                Replay_Id__c = event.ReplayId,
                Error_Message__c = e.getMessage(),
                Event_Payload__c = JSON.serialize(event),
                Occurred_At__c = DateTime.now()
            );
            insert as user errorLog;
        }
    }
}

Best Practices

Platform Events

1. Batch Publishing: Publish events in batches (up to 2,000 per transaction)
2. Idempotent Subscribers: Design subscribers to handle duplicate events (especially for high-volume)
3. ReplayId Tracking: Store ReplayIds for resume capability
4. Error Isolation: Catch exceptions in subscriber loops to process remaining events
5. Resume Checkpoints: Use setResumeCheckpoint() for high-volume events

Change Data Capture

1. Field Filtering: Only process relevant field changes to reduce processing overhead
2. Async Processing: Use Queueable/Future for callouts or long-running operations
3. ChangeType Routing: Use switch statements for different change types
4. RecordIds Array: Handle multiple RecordIds (CDC can batch related changes)
5. Commit Metadata: Use getCommitTimestamp() and getCommitUser() for audit trails

Governor Limits

Testing Event-Driven Integrations

Platform Event Test

@isTest
private class OrderEventTest {

    @isTest
    static void testEventPublish() {
        Test.startTest();

        Order_Status__e event = new Order_Status__e(
            RecordId__c = '006xx000000XXXAAA',
            Status__c = 'Completed'
        );
        Database.SaveResult sr = EventBus.publish(event);

        Test.stopTest();

        System.assert(sr.isSuccess(), 'Event should publish successfully');
    }

    @isTest
    static void testEventSubscriber() {
        // Subscriber triggers execute synchronously in tests
        Test.startTest();

        Order_Status__e event = new Order_Status__e(
            RecordId__c = '006xx000000XXXAAA',
            Status__c = 'Completed'
        );
        EventBus.publish(event);

        Test.stopTest();

        // Verify subscriber logic executed
        // (check that handler updated related records)
    }
}

CDC Test

@isTest
private class AccountCDCTest {

    @isTest
    static void testAccountUpdate() {
        Account acc = new Account(Name = 'Test Account');
        insert acc;

        Test.startTest();

        acc.BillingCity = 'San Francisco';
        update acc;

        Test.stopTest();

        // CDC events don't fire in test context
        // Must test handler methods directly
        AccountChangeEvent mockEvent = new AccountChangeEvent();
        // Note: Can't instantiate ChangeEvent in Apex
        // Test handler logic with mock data instead
    }
}

Related Resources

• Callout Patterns - REST and SOAP callout implementations
• Main Skill Documentation - sf-integration overview
• Platform Event Templates - Event definitions and triggers
• CDC Templates - Change Data Capture triggers