Afficher en direct les changements Salesforce avec Mercure

Le scénario

Chez Acme Assurances (données fictives), les gestionnaires traitent les sinistres dans Salesforce, sur l’objet Sinistre__c de l’org acme-demo. Les assurés suivent leur dossier dans un espace client Symfony. Quand un gestionnaire passe un dossier à « Indemnisé », l’assuré qui a la page ouverte doit le voir aussitôt, sans rafraîchir.

Les briques existent déjà dans les articles précédents : Change Data Capture et la Pub/Sub API pour savoir ce qui change dans Salesforce, Mercure pour pousser au navigateur, et les mises à jour privées pour que chacun ne voie que ses dossiers. Reste à les relier.

L’architecture du relais

1. Salesforce publie un événement CDC à chaque modification de Sinistre__c.
2. Un abonné Pub/Sub, processus permanent, reçoit l’événement et le transmet à Symfony par une route interne.
3. Symfony met le changement en file Messenger, puis un worker le traduit : filtrage des champs, identification de l’assuré, construction des topics.
4. Le hub Mercure remet la mise à jour privée aux navigateurs autorisés.

Le navigateur ne parle jamais à Salesforce et ne reçoit aucun jeton de l’org. Toute la logique métier (qui voit quoi, quels champs sont publics) vit à un seul endroit, dans Symfony, où elle se teste unitairement.

L’abonné Pub/Sub transmet

L’abonné de l’article Pub/Sub API, en Python avec gRPC, reste un simple tuyau : il filtre l’objet suivi et poste l’événement décodé.

import os, requests

RELAIS_URL   = os.environ["RELAIS_URL"]      # https://espace.acme-assurances.example/internal/salesforce-changes
RELAIS_TOKEN = os.environ["RELAIS_TOKEN"]

def handle_change(change: dict) -> None:
    header = change["ChangeEventHeader"]
    if header["entityName"] != "Sinistre__c":
        return

    # L'abonné reste simple : il transmet l'en-tête et les champs, Symfony décide de la suite.
    response = requests.post(
        RELAIS_URL,
        json={
            "transactionKey": header["transactionKey"],
            "changeType":     header["changeType"],
            "recordIds":      header["recordIds"],
            "fields":         {k: v for k, v in change.items() if k != "ChangeEventHeader"},
        },
        headers={"Authorization": f"Bearer {RELAIS_TOKEN}"},
        timeout=5,
    )
    response.raise_for_status()   # en cas d'échec, le replay_id n'est pas avancé : l'événement sera relu

Le point clé est l’ordre des opérations : le replay_id n’est mémorisé qu’après une réponse positive de Symfony. Si l’application est indisponible, l’abonné rejoue l’événement au redémarrage, dans la limite de rétention de trois jours.

Symfony reçoit et met en file

La route interne valide le jeton, découpe l’événement par enregistrement et répond 202 immédiatement :

<?php

namespace App\Controller\Internal;

use App\Message\SalesforceRecordChanged;
use Symfony\Component\DependencyInjection\Attribute\Autowire;
use Symfony\Component\HttpFoundation\JsonResponse;
use Symfony\Component\HttpFoundation\Request;
use Symfony\Component\HttpFoundation\Response;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Routing\Attribute\Route;

final class SalesforceChangesController
{
    public function __construct(
        private readonly MessageBusInterface $bus,
        #[Autowire('%env(RELAIS_TOKEN)%')] private readonly string $token,
    ) {
    }

    #[Route('/internal/salesforce-changes', methods: ['POST'])]
    public function __invoke(Request $request): Response
    {
        // Route interne : jeton partagé comparé en temps constant, et filtrage réseau en amont.
        $bearer = substr($request->headers->get('Authorization', ''), 7);
        if (!hash_equals($this->token, $bearer)) {
            return new Response('', Response::HTTP_UNAUTHORIZED);
        }

        $change = $request->toArray();
        foreach ($change['recordIds'] as $recordId) {
            $this->bus->dispatch(new SalesforceRecordChanged(
                $change['transactionKey'],
                $change['changeType'],
                $recordId,
                $change['fields'],
            ));
        }

        return new JsonResponse(null, Response::HTTP_ACCEPTED);
    }
}

Ce découpage compte : une modification en masse dans Salesforce produit un seul événement CDC pour des dizaines d’enregistrements, chacun appartenant à un assuré différent.

Traduire un changement en mise à jour

Le handler applique une liste blanche de champs, relit ce que l’événement ne contient pas, puis publie une mise à jour privée sur deux topics : celui de l’assuré et celui de son agence.

<?php

namespace App\MessageHandler;

use App\Message\SalesforceRecordChanged;
use App\Salesforce\SalesforceClient;
use Symfony\Component\Mercure\HubInterface;
use Symfony\Component\Mercure\Update;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
final class PublishSinistreChange
{
    // Liste blanche : seuls ces champs peuvent atteindre un navigateur.
    private const VISIBLE = ['Statut__c' => 'statut', 'Montant_Indemnise__c' => 'montantIndemnise'];

    public function __construct(
        private readonly SalesforceClient $salesforce,
        private readonly HubInterface $hub,
    ) {
    }

    public function __invoke(SalesforceRecordChanged $change): void
    {
        $visible = array_intersect_key($change->fields, self::VISIBLE);
        if ($visible === [] && $change->changeType === 'UPDATE') {
            return;   // changement interne (note de gestionnaire, champ technique) : rien à diffuser
        }

        // Une modification CDC ne contient que les champs changés : on relit l'assuré et la référence.
        $sinistre = $this->salesforce->get('Sinistre__c', $change->recordId, ['Name', 'Assure_Id__c', 'Agence__c']);
        $base = 'https://espace.acme-assurances.example';

        $data = ['reference' => $sinistre['Name'], 'changeType' => $change->changeType];
        foreach ($visible as $field => $value) {
            $data[self::VISIBLE[$field]] = $value;
        }

        $this->hub->publish(new Update(
            topics: [
                "$base/assures/{$sinistre['Assure_Id__c']}/sinistres/{$sinistre['Name']}",
                "$base/agences/{$sinistre['Agence__c']}/sinistres/{$sinistre['Name']}",
            ],
            data: json_encode($data, JSON_THROW_ON_ERROR),
            private: true,
            // Identifiant stable : un même changement relu deux fois porte le même id.
            id: $change->transactionKey . ':' . $change->recordId,
            type: 'sinistre.change',
        ));
    }
}

La relecture par le client Salesforce du middleware coûte un appel d’API par changement diffusé. Si le volume devient important, on garde en base locale la correspondance sinistre → assuré, alimentée par les créations CDC.

Côté navigateur

Le contrôleur Stimulus de l’article Mercure s’utilise tel quel : la page rend l’état courant, l’abonnement met à jour les champs présents dans la charge.

{% set topic = base ~ '/sinistres/' ~ sinistre.reference %}
<section data-controller="live"
         data-live-url-value="{{ mercure(topic, { subscribe: topic })|escape('html_attr') }}"
         data-live-type-value="sinistre.change">
    <p>Statut : <strong data-live-target="field" data-field="statut">{{ sinistre.statut }}</strong></p>
    <p>Indemnité : <span data-live-target="field" data-field="montantIndemnise">{{ sinistre.montantIndemnise }}</span> €</p>
</section>

Le cookie d’abonnement, posé avec la page, n’autorise que les topics de l’assuré connecté. Le gestionnaire, sur son tableau de bord Symfony, s’abonne au gabarit de son agence et voit défiler les dossiers traités par ses collègues.

Doublons, trous et reconnexions

• Doublons : après une reprise, l’abonné peut retransmettre un événement déjà traité. Mettre à jour un affichage deux fois avec la même valeur est sans effet ; l’identifiant stable transactionKey:recordId permet en plus au navigateur de les ignorer.
• Événements de trou : quand Salesforce ne peut pas fournir le détail d’un changement, il émet un événement GAP_…. On publie alors une simple invalidation et la page relit le dossier par l’API.
• Reconnexions : le navigateur se reconnecte seul au hub. Si le hub conserve un historique, Last-Event-ID rejoue les mises à jour manquées ; sinon, la page relit l’état à la reconnexion.

# Abonné Python : sur un événement de trou, demander une invalidation plutôt qu'un détail.
if header["changeType"].startswith("GAP_"):
    requests.post(RELAIS_URL, json={"changeType": "GAP", "recordIds": header["recordIds"],
                                    "transactionKey": header["transactionKey"], "fields": {}},
                  headers={"Authorization": f"Bearer {RELAIS_TOKEN}"}, timeout=5)
    return

Bonnes pratiques

• Liste blanche de champs : une note interne de gestionnaire ou un indicateur de fraude ne doit jamais atteindre un navigateur, même privé.
• Éviter l’écho : un changement fait par l’assuré depuis l’espace client revient par CDC ; changeOrigin ou l’utilisateur d’intégration permettent de le reconnaître si l’interface l’a déjà affiché.
• Surveiller le retard : mesurer l’écart entre commitTimestamp et la publication Mercure ; au-delà de quelques secondes, la file ou l’abonné sature.
• Garder le temps réel optionnel : si le hub tombe, la page reste juste au rafraîchissement suivant. Mercure améliore l’expérience, il ne porte pas la donnée.