Pub/Sub API et Change Data Capture : suivre Salesforce en temps réel

Pourquoi l’événementiel

Pour garder une application externe à jour avec Salesforce, la solution naïve consiste à interroger l’API toutes les cinq minutes : « qu’est-ce qui a changé depuis la dernière fois ? ». Cela consomme des appels même quand rien ne change, introduit un délai, et rate les suppressions.

L’approche événementielle inverse le sens : Salesforce publie chaque changement, et l’application s’abonne. Fil rouge : une application de gestion fictive d’Acme Assurances qui doit refléter en temps réel les contacts de l’org acme-demo.

Pour un chargement initial ou un rattrapage massif, la Bulk API 2.0 reste l’outil adapté ; l’événementiel prend le relais pour le flux continu.

Change Data Capture

Change Data Capture (CDC) publie un événement à chaque création, modification, suppression ou restauration d’un enregistrement. On l’active objet par objet dans Setup → Change Data Capture ; chaque objet suivi dispose alors d’un canal, par exemple /data/ContactChangeEvent.

{
  "ChangeEventHeader": {
    "entityName": "Contact",
    "recordIds": ["003XXXXXXXXXXXXXXX"],
    "changeType": "UPDATE",
    "changeOrigin": "com/salesforce/api/rest/62.0",
    "transactionKey": "0002f3a1-...",
    "sequenceNumber": 1,
    "commitTimestamp": 1793088000000,
    "commitUser": "005XXXXXXXXXXXXXXX",
    "changedFields": ["0x0C00"]
  },
  "Email": "camille.martin@example.com",
  "Phone": "+33 1 23 45 67 89",
  "LastModifiedDate": 1793088000000
}

• Une modification ne contient que les champs modifiés, plus l’en-tête ; une création contient tous les champs renseignés.
• recordIds peut contenir plusieurs identifiants : une même modification appliquée à plusieurs enregistrements produit un seul événement.
• Avec la Pub/Sub API, changedFields arrive sous forme de bitmap à décoder avec le schéma ; les bibliothèques d’exemple de Salesforce fournissent cette fonction.

Pour des événements métier (« contrat signé ») plutôt que des changements de données, on utilise des Platform Events (/event/Contrat_Signe__e), consommés exactement de la même façon.

La Pub/Sub API

La Pub/Sub API est l’API de Salesforce pour publier et consommer ces événements. Elle repose sur gRPC (HTTP/2) et encode les messages en Avro, un format binaire compact décrit par un schéma.

• Subscribe : flux bidirectionnel ; le client demande N événements, le serveur les envoie au fil de l’eau.
• GetSchema : renvoie le schéma Avro correspondant à l’identifiant porté par chaque événement.
• Publish : publie des Platform Events.

L’authentification réutilise un jeton OAuth (voir Authentifier une intégration Salesforce avec OAuth 2.0), passé dans les métadonnées gRPC avec l’URL de l’instance et l’Id de l’org. Les stubs client se génèrent depuis le fichier pubsub_api.proto publié par Salesforce, dans le langage de son choix.

S’abonner en Python

Esquisse d’un abonné, avec les stubs générés et la bibliothèque avro :

import io, threading
import avro.io, avro.schema, grpc
import pubsub_api_pb2 as pb2            # générés depuis pubsub_api.proto
import pubsub_api_pb2_grpc as pb2_grpc

TOPIC = "/data/ContactChangeEvent"
BATCH = 100

token = salesforce_token()              # jeton OAuth de l'utilisateur d'intégration
metadata = (
    ("accesstoken", token["access_token"]),
    ("instanceurl", token["instance_url"]),
    ("tenantid", ORG_ID),               # Id de l'org (00D...)
)
channel = grpc.secure_channel("api.pubsub.salesforce.com:7443", grpc.ssl_channel_credentials())
stub = pb2_grpc.PubSubStub(channel)

schemas = {}
def decode(event):
    schema_id = event.event.schema_id
    if schema_id not in schemas:
        info = stub.GetSchema(pb2.SchemaRequest(schema_id=schema_id), metadata=metadata)
        schemas[schema_id] = avro.schema.parse(info.schema_json)
    reader = avro.io.DatumReader(schemas[schema_id])
    return reader.read(avro.io.BinaryDecoder(io.BytesIO(event.event.payload)))

# Contrôle de flux : on ne redemande des événements qu'après avoir traité le lot précédent.
ready = threading.Semaphore(1)
def fetch_requests(last_replay_id):
    first = True
    while True:
        ready.acquire()
        request = pb2.FetchRequest(topic_name=TOPIC, num_requested=BATCH)
        if first and last_replay_id:
            request.replay_preset = pb2.ReplayPreset.CUSTOM
            request.replay_id = last_replay_id
        first = False
        yield request

for response in stub.Subscribe(fetch_requests(load_replay_id()), metadata=metadata):
    for event in response.events:
        handle_change(decode(event))
        save_replay_id(event.replay_id)  # après traitement : au pire, un événement est rejoué
    if response.pending_num_requested == 0:
        ready.release()

Le contrôle de flux est explicite : le serveur n’envoie jamais plus que ce qui a été demandé. On redemande un lot quand le précédent est traité, ce qui protège l’application en cas de pic. Les réponses sans événement servent de signal de vie et gardent le flux ouvert.

Reprendre là où on s’est arrêté

Chaque événement porte un replay ID, opaque et croissant. L’abonné l’enregistre (base, fichier) après chaque traitement, et le fournit au redémarrage avec ReplayPreset.CUSTOM : il reçoit alors tout ce qu’il a manqué.

Les événements sont conservés trois jours. Une panne plus longue, ou un replay ID trop ancien, impose une resynchronisation complète (par la Bulk API) avant de reprendre l’abonnement en LATEST.

Comme le replay ID est enregistré après le traitement, un arrêt brutal peut faire rejouer un événement : le traitement doit être idempotent (un upsert par Id, pas une insertion aveugle).

Événements de trou et resynchronisation

Quand Salesforce ne peut pas produire le détail d’un changement (transaction très volumineuse, incident interne), il publie un événement de trou : changeType commence par GAP_ et seuls les identifiants sont fournis. Un volume extrême produit des événements GAP_OVERFLOW. Dans tous les cas, la parade est la même : relire les enregistrements concernés à la source.

def handle_change(change):
    header = change["ChangeEventHeader"]
    change_type = header["changeType"]

    if change_type.startswith("GAP_"):
        # Détail perdu (transaction trop volumineuse, incident) : relire l'enregistrement à la source.
        for record_id in header["recordIds"]:
            resync_from_rest_api(header["entityName"], record_id)
        return

    for record_id in header["recordIds"]:
        if change_type in ("CREATE", "UPDATE", "UNDELETE"):
            upsert_local(header["entityName"], record_id, change)
        elif change_type == "DELETE":
            delete_local(header["entityName"], record_id)

Bonnes pratiques

• Un seul consommateur actif par flux, ou un partage explicite du travail : deux abonnés indépendants traitent chacun tous les événements.
• Boucles d’écho : si l’application écrit aussi dans Salesforce, ignorer les événements dont commitUser est son propre utilisateur d’intégration.
• Ordre : l’ordre est garanti au sein d’un canal ; pour des données liées (comptes et contacts), suivre un canal combiné ou tolérer un parent absent quelques instants.
• Limites : le nombre d’événements livrés par jour dépend de l’édition et des licences ; le suivre dans /limits et ne suivre que les objets utiles.
• Enrichissement : CDC peut inclure des champs supplémentaires (enriched fields) pour éviter un appel REST par événement, par exemple l’identifiant externe.