# Event-Driven Integration Patterns
This document provides detailed implementation patterns for Platform Events and Change Data Capture (CDC) in Salesforce integrations.
> **Parent Document**: [integration-connectivity-generate/SKILL.md](../SKILL.md)
> **Related**: [callout-patterns.md](./callout-patterns.md)
---
## Table of Contents
- [Platform Events](#platform-events)
- [Platform Event Definition](#platform-event-definition)
- [Event Publisher](#event-publisher)
- [Event Subscriber Trigger](#event-subscriber-trigger)
- [High-Volume vs Standard-Volume Events](#high-volume-vs-standard-volume-events)
- [Change Data Capture (CDC)](#change-data-capture-cdc)
- [CDC Enablement](#cdc-enablement)
- [CDC Subscriber Trigger](#cdc-subscriber-trigger)
- [CDC Handler Service](#cdc-handler-service)
- [Field-Specific Change Detection](#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**: `assets/platform-events/platform-event-definition.object-meta.xml`
#### Standard Volume Event
Best for moderate event volumes (~2,000 events/hour):
```xml
Deployed
StandardVolume
{{EventPluralLabel}}
PublishAfterCommit
{{FieldName}}__c
Text
255
RecordId__c
Text
18
Salesforce record ID related to this event
Timestamp__c
DateTime
When the event was triggered
```
#### High-Volume Event
Best for millions of events per day:
```xml
Deployed
HighVolume
{{EventPluralLabel}}
PublishAfterCommit
{{FieldName}}__c
Text
255
```
#### Key Configuration Options
| Option | Values | Description |
|--------|--------|-------------|
| **eventType** | `StandardVolume`, `HighVolume` | Event throughput capacity |
| **publishBehavior** | `PublishAfterCommit`, `PublishImmediately` | When events are published |
| **deploymentStatus** | `Deployed`, `InDevelopment` | Deployment status |
**PublishBehavior Details**:
- `PublishAfterCommit`: Event published only if transaction commits (recommended)
- `PublishImmediately`: Event published immediately, even if transaction rolls back
---
### Event Publisher
**Template**: `assets/platform-events/event-publisher.cls`
#### Bulk Event Publisher
```apex
public with sharing class {{EventName}}Publisher {
public static void publishEvents(List<{{EventName}}__e> events) {
if (events == null || events.isEmpty()) {
return;
}
List 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 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**:
```apex
Map eventData = new Map{
'recordId' => '001xx000003DXXXAAA',
'status' => 'Completed',
'amount' => 1000.00
};
OrderStatusPublisher.publishSingleEvent(eventData);
```
**Bulk Events**:
```apex
List events = new List();
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**: `assets/platform-events/event-subscriber-trigger.trigger`
**Template**: `assets/platform-events/event-subscriber-action.cls`
#### Standard Volume Subscriber
```apex
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)
```apex
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
```apex
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 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 payload = new Map{
'recordId' => recordId,
'status' => status
};
System.enqueueJob(new ExternalSyncQueueable(payload));
}
}
```
---
### High-Volume vs Standard-Volume Events
| Feature | Standard-Volume | High-Volume |
|---------|----------------|-------------|
| **Throughput** | ~2,000 events/hour | Millions/day |
| **Delivery** | Exactly-once | At-least-once (may deliver duplicates) |
| **Retention** | 3 days (72 hours) | 24 hours |
| **Replay** | ReplayId from last 3 days | ReplayId from last 24 hours |
| **Use Case** | Low-volume integrations, workflows | IoT, real-time analytics, high-traffic |
| **Cost** | Included in platform | Additional licensing |
#### 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
Account
true
```
#### 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**: `assets/cdc/cdc-subscriber-trigger.trigger`
#### Basic CDC Subscriber
```apex
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 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
| Field | Type | Description |
|-------|------|-------------|
| `getChangeType()` | String | CREATE, UPDATE, DELETE, UNDELETE |
| `getRecordIds()` | List | Record IDs affected (usually 1, up to 5 for related changes) |
| `getChangedFields()` | List | Field API names that changed (UPDATE only) |
| `getCommitTimestamp()` | Long | Transaction commit timestamp |
| `getCommitUser()` | String | User ID who made the change |
| `getCommitNumber()` | Long | Monotonically increasing commit number |
| `getEntityName()` | String | Object API name |
---
### CDC Handler Service
**Template**: `assets/cdc/cdc-handler.cls`
```apex
public with sharing class {{ObjectName}}CDCHandler {
public static void handleCreate({{ObjectName}}ChangeEvent event) {
// Sync to external system on create
Map payload = buildPayload(event);
System.enqueueJob(new ExternalSystemSyncQueueable(payload, 'CREATE'));
}
public static void handleUpdate({{ObjectName}}ChangeEvent event, List changedFields) {
// Only sync if relevant fields changed
Set fieldsToWatch = new Set{'Name', 'Status__c', 'Amount__c'};
Boolean relevantChange = false;
for (String field : changedFields) {
if (fieldsToWatch.contains(field)) {
relevantChange = true;
break;
}
}
if (relevantChange) {
Map payload = buildPayload(event);
payload.put('changedFields', changedFields);
System.enqueueJob(new ExternalSystemSyncQueueable(payload, 'UPDATE'));
}
}
public static void handleDelete(String recordId) {
Map payload = new Map{'recordId' => recordId};
System.enqueueJob(new ExternalSystemSyncQueueable(payload, 'DELETE'));
}
public static void handleUndelete({{ObjectName}}ChangeEvent event) {
handleCreate(event); // Treat undelete like create
}
private static Map buildPayload({{ObjectName}}ChangeEvent event) {
return new Map{
'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
```apex
public static void handleUpdate(AccountChangeEvent event, List changedFields) {
// Only process if billing address changed
Set billingFields = new Set{
'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
```apex
public static void handleUpdate(OpportunityChangeEvent event, List changedFields) {
Set changedFieldSet = new Set(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
| Use Case | Platform Events | Change Data Capture |
|----------|----------------|---------------------|
| **Custom business events** | **Preferred** | Not applicable |
| **Record change notifications** | Requires custom trigger | **Automatic** (no code) |
| **External system sync** | Both work | **CDC** (lower maintenance) |
| **Custom event fields** | Fully customizable | Limited to object fields |
| **Event filtering** | Filter in publisher code | Filter in subscriber code |
| **Performance overhead** | Manual event creation | Automatic (minimal overhead) |
### Decision Matrix
```text
┌───────────────────────────────────────────────────────────────────────┐
│ 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
```apex
public with sharing class AccountCDCHandler {
public static void handleUpdate(AccountChangeEvent event, List changedFields) {
// Extract data
String recordId = event.ChangeEventHeader.getRecordIds()[0];
String accountName = event.Name;
// Queue async callout to external CRM
Map payload = new Map{
'salesforceId' => recordId,
'name' => accountName,
'changedFields' => changedFields,
'timestamp' => event.ChangeEventHeader.getCommitTimestamp()
};
System.enqueueJob(new CRMSyncQueueable(payload));
}
}
```
### Event Replay with Stored ReplayId
```apex
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
```apex
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
| Limit | Value | Notes |
|-------|-------|-------|
| Platform Events published/transaction | 2,000 | Both Standard and High-Volume |
| Platform Event delivery | Asynchronous | Delivered to subscribers after commit |
| CDC events/hour (per object) | 250,000 | Auto-throttled if exceeded |
| Event message size | 1 MB | Total size of all fields |
| Event retention (Standard) | 3 days | ReplayId available for 72 hours |
| Event retention (High-Volume) | 24 hours | ReplayId available for 24 hours |
---
## Testing Event-Driven Integrations
### Platform Event Test
```apex
@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
```apex
@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
}
}
```
---
## Pub/Sub API (Recommended for External Consumers)
The Pub/Sub API is the **recommended mechanism** for external systems subscribing to Platform Events and CDC events. It replaces the legacy Streaming API (CometD).
### Why Pub/Sub API
| Feature | Pub/Sub API | Legacy Streaming API |
|---------|-------------|---------------------|
| **Protocol** | gRPC (binary, high performance) | CometD (long-polling, HTTP overhead) |
| **Authentication** | OAuth 2.0 | Session-based |
| **Event Types** | Platform Events, CDC, Custom Channels | PushTopic, Generic Events |
| **Status** | **Current** | **Deprecated — no new investments** |
### Subscription Modes
- **Subscribe**: Stream events from a given replay ID forward
- **PublishStream**: Bi-directional — publish events via gRPC
- **ManagedSubscribe**: Salesforce manages replay state (simplest)
### LWC Internal Subscription
For Lightning Web Components, use the `empApi` module:
```javascript
import { subscribe, unsubscribe, onError } from 'lightning/empApi';
connectedCallback() {
subscribe('/event/Order_Status__e', -1, (response) => {
this.handleEvent(response.data.payload);
}).then((sub) => { this.subscription = sub; });
}
disconnectedCallback() {
unsubscribe(this.subscription);
}
```
> **Note**: `empApi` uses CometD internally but is the supported LWC API. For external consumers, always use the gRPC-based Pub/Sub API.
---
## Platform Event Anti-Patterns
### 1. Publishing from Trigger on Same Event Object → Infinite Loop
```apex
// ❌ WRONG: Trigger on Order_Status__e publishes Order_Status__e
trigger OrderStatusSubscriber on Order_Status__e (after insert) {
for (Order_Status__e event : Trigger.new) {
// This creates an infinite loop!
EventBus.publish(new Order_Status__e(Status__c = 'Processed'));
}
}
```
**Fix**: Never publish the same event type from its own subscriber trigger. Use a different event type or update a record instead.
### 2. PublishImmediately When Data Integrity Matters
```apex
// ❌ WRONG: Event fires even if the transaction rolls back
Order_Status__e event = new Order_Status__e();
event.Status__c = 'Completed';
// publishBehavior = PublishImmediately in event definition
EventBus.publish(event);
// If subsequent DML fails, the event was already published
// External system thinks order is "Completed" but it's not
```
**Fix**: Use `PublishAfterCommit` (the default) when the event represents a state change that depends on the transaction succeeding.
### 3. Using Platform Events for Synchronous-Style Flow Orchestration
```javascript
// ❌ WRONG: Using events to simulate synchronous request/response
Flow A → Publish "Request" Event → Subscriber triggers Flow B → Publish "Response" Event → ???
```
**Fix**: Platform Events are asynchronous and unordered. For synchronous orchestration, use Subflows, @InvocableMethod, or direct Apex calls. Events are for decoupled, fire-and-forget communication.
### 4. Oversized Event Payloads
> **1 MB message size limit.** Balance payload size — smaller = faster delivery, larger = fewer API calls. Include record IDs and essential context; let consumers query Salesforce for full records if needed.
---
## Related Resources
- [Callout Patterns](./callout-patterns.md) - REST and SOAP callout implementations
- [Event-Driven Architecture Guide](./event-driven-architecture-guide.md) - EDA patterns, Pub/Sub API deep dive, monitoring
- [Main Skill Documentation](../SKILL.md) - integration-connectivity-generate overview
- [Platform Event Templates](../assets/platform-events/) - Event definitions and triggers
- [CDC Templates](../assets/cdc/) - Change Data Capture triggers