Synchroniser vers Salesforce avec Symfony Messenger

Pourquoi une file de messages

Dans un middleware Symfony, un appel synchrone à Salesforce lie le sort de l’appelant à celui de l’org : maintenance planifiée, verrou sur un compte très sollicité, quota d’API atteint ou simple lenteur, et la requête d’origine échoue.

Pour une écriture qui n’attend pas de réponse immédiate (une fiche client modifiée dans le portail, un contrat signé), une file de messages découple les deux mondes :

• l’appelant reçoit un 202 Accepted en quelques millisecondes ;
• un worker envoie à Salesforce à son rythme, regroupe les écritures et réessaie en cas d’incident passager ;
• un message impossible à traiter est mis de côté, sans bloquer les suivants.

Symfony Messenger fournit tout cela : transports, stratégie de reprise, transport d’échec et traitement par lots.

Le message et le transport

Le message contient les valeurs à écrire, prêtes pour Salesforce, et une version de la fiche source :

<?php

namespace App\Message;

// Un message transporte des données, pas d'entité Doctrine : il sera sérialisé dans la file.
final readonly class SyncClient
{
    public function __construct(
        public string $ref,          // CLI-000042
        public array $fields,        // ['FirstName' => ..., 'LastName' => ..., 'Email' => ...]
        public int $version,         // version de la fiche source, pour ignorer un message périmé
    ) {
    }
}

Un transport dédié isole la synchronisation Salesforce des autres tâches asynchrones et porte sa propre stratégie de reprise. Le délai croît à chaque tentative : un quota épuisé a le temps de se reconstituer.

# config/packages/messenger.yaml
framework:
    messenger:
        failure_transport: failed
        transports:
            salesforce:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'   # doctrine://default?queue_name=salesforce
                retry_strategy:
                    max_retries: 5
                    delay: 30000        # 30 s, puis 1 min 30, 4 min 30…
                    multiplier: 3
                    max_delay: 3600000  # plafonné à 1 h
            failed: 'doctrine://default?queue_name=failed'
        routing:
            App\Message\SyncClient: salesforce

// Dans le contrôleur du middleware : on accepte, on enfile, on répond tout de suite.
$this->bus->dispatch(new SyncClient($ref, $input->toSalesforceFields(), $input->version));

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

Un handler idempotent

Messenger garantit qu’un message sera traité au moins une fois, pas exactement une fois : un worker interrompu après l’appel Salesforce mais avant l’acquittement rejouera le message. L’upsert par identifiant externe rend ce rejeu inoffensif.

<?php

namespace App\MessageHandler;

use App\Message\SyncClient;
use App\Salesforce\SalesforceClient;
use App\Salesforce\SalesforceException;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
use Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException;

#[AsMessageHandler]
final class SyncClientHandler
{
    public function __construct(private readonly SalesforceClient $salesforce)
    {
    }

    public function __invoke(SyncClient $message): void
    {
        try {
            // Upsert : créer ou mettre à jour selon l'External ID. Rejouer le message ne crée pas de doublon.
            $this->salesforce->request('PATCH', '/sobjects/Contact/Ref_Externe__c/' . rawurlencode($message->ref), [
                'json' => $message->fields + ['Version_Source__c' => $message->version],
            ]);
        } catch (SalesforceException $e) {
            if (self::isDataError($e)) {
                // Réessayer ne corrigera pas un e-mail invalide : droit au transport d'échec.
                throw new UnrecoverableMessageHandlingException($e->getMessage(), 0, $e);
            }
            throw $e; // verrou, quota, indisponibilité : la stratégie de reprise s'applique
        }
    }

    private static function isDataError(SalesforceException $e): bool
    {
        return in_array($e->errorCode, [
            'INVALID_EMAIL_ADDRESS', 'REQUIRED_FIELD_MISSING', 'FIELD_CUSTOM_VALIDATION_EXCEPTION',
            'STRING_TOO_LONG', 'DUPLICATES_DETECTED',
        ], true);
    }
}

Reste l’ordre : deux modifications rapprochées d’une même fiche peuvent arriver inversées après une reprise. Le champ fictif Version_Source__c permet à une règle de validation, ou à un trigger, de refuser une version plus ancienne que celle déjà enregistrée.

Regrouper par lots de 200

Un appel par message consomme vite le quota. Une ressource sObject Collections accepte jusqu’à 200 enregistrements par appel, upsert compris :

PATCH /services/data/v62.0/composite/sobjects/Contact/Ref_Externe__c
{ "allOrNone": false, "records": [
    { "attributes": { "type": "Contact" }, "Ref_Externe__c": "CLI-000001", "LastName": "Martin" },
    { "attributes": { "type": "Contact" }, "Ref_Externe__c": "CLI-000002", "LastName": "Bernard", "Email": "yanis@" }
] }

→ [ { "id": "003...", "success": true, "created": false, "errors": [] },
    { "success": false, "errors": [ { "statusCode": "INVALID_EMAIL_ADDRESS", "message": "Email: invalid email address" } ] } ]

Un handler de lot Messenger accumule les messages et appelle process() à 200, ou plus tôt quand la file se vide. Chaque message est acquitté ou rejeté individuellement selon sa ligne de résultat. Ce handler remplace le précédent : deux handlers déclarés pour le même message s’exécuteraient tous les deux.

<?php

namespace App\MessageHandler;

use App\Message\SyncClient;
use App\Salesforce\SalesforceClient;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
use Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException;
use Symfony\Component\Messenger\Handler\Acknowledger;
use Symfony\Component\Messenger\Handler\BatchHandlerInterface;
use Symfony\Component\Messenger\Handler\BatchHandlerTrait;

#[AsMessageHandler]
final class SyncClientBatchHandler implements BatchHandlerInterface
{
    use BatchHandlerTrait;

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

    public function __invoke(SyncClient $message, ?Acknowledger $ack = null): mixed
    {
        return $this->handle($message, $ack);
    }

    /** @param list<array{SyncClient, Acknowledger}> $jobs */
    private function process(array $jobs): void
    {
        $records = array_map(static fn (array $job): array => [
            'attributes' => ['type' => 'Contact'],
            'Ref_Externe__c' => $job[0]->ref,
            'Version_Source__c' => $job[0]->version,
        ] + $job[0]->fields, $jobs);

        // allOrNone false : un enregistrement refusé n'annule pas les 199 autres.
        $results = $this->salesforce->request('PATCH', '/composite/sobjects/Contact/Ref_Externe__c', [
            'json' => ['allOrNone' => false, 'records' => $records],
        ]);

        // La réponse suit l'ordre de la requête : un résultat par message.
        foreach ($jobs as $i => [, $ack]) {
            $result = $results[$i];
            $result['success']
                ? $ack->ack($result['id'])
                : $ack->nack(new UnrecoverableMessageHandlingException($result['errors'][0]['message'] ?? 'Refusé'));
        }
    }

    // 200 : maximum d'enregistrements par appel sObject Collections.
    private function getBatchSize(): int
    {
        return 200;
    }
}

Une exception levée par l’appel lui-même (quota, indisponibilité) rejette tout le lot, et chaque message repart dans la stratégie de reprise. On divise ainsi le nombre d’appels par 200 sans perdre le suivi individuel. Au-delà de quelques milliers d’enregistrements par passage, la Bulk API prend le relais.

Réessayer ou abandonner

Tout l’intérêt de la file repose sur une bonne distinction entre erreurs passagères et erreurs définitives :

Sans cette distinction, un message invalide fait cinq tentatives inutiles, et une panne passagère d’une heure épuise les reprises de tous les messages en attente. Le plafond max_delay et le nombre de tentatives se calibrent sur la durée d’une maintenance Salesforce.

Exploiter les workers

# Un worker redémarré régulièrement : il libère la mémoire et charge le nouveau code après un déploiement.
php bin/console messenger:consume salesforce --time-limit=3600 --memory-limit=256M -vv

# Au déploiement : demander aux workers de s'arrêter proprement (le superviseur les relance).
php bin/console messenger:stop-workers

# Messages abandonnés : inspecter, puis rejouer après correction.
php bin/console messenger:failed:show
php bin/console messenger:failed:retry

En production, un superviseur maintient les workers en vie. Deux processus suffisent souvent : au-delà, le gain est limité par les verrous Salesforce sur les enregistrements parents.

; /etc/supervisor/conf.d/messenger-salesforce.conf
[program:messenger-salesforce]
command=php /srv/app/bin/console messenger:consume salesforce --time-limit=3600 --memory-limit=256M
user=www-data
numprocs=2
process_name=%(program_name)s_%(process_num)02d
autostart=true
autorestart=true

Bonnes pratiques

• Surveiller la file : nombre de messages en attente et taille du transport d’échec, avec une alerte au-delà d’un seuil.
• Messages légers : des valeurs simples, jamais une entité ni un objet de service ; un changement de classe ne doit pas rendre illisibles les messages déjà en file.
• Lots et verrous : regrouper des contacts d’un même compte dans un lot limite les conflits de verrou entre workers.
• Transport fiable : Doctrine, Redis ou AMQP ; jamais sync:// en production pour ce cas, qui supprimerait tout l’intérêt de la file.
• Tests : le transport in-memory:// vérifie qu’un contrôleur publie le bon message ; le handler se teste seul avec MockHttpClient.