RabbitMQعلى تدفّق موجود.

$this->publisher->publish(
    'vendor.order.sync',
    json_encode($order->getData())  // ثقيل وقديم
);

$message = json_encode([
    'order_id'  => (int) $order->getId(),
    'increment' => $order->getIncrementId(),  // للتتبّع
    'store_id'  => (int) $order->getStoreId(),
    'event'     => 'order_placed',                 // للتمييز
    'queued_at' => gmdate('c'),                   // للتشخيص
]);

$this->publisher->publish('vendor.order.sync', $message);

<?xml version="1.0"?>
<schema xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
  <table name="vendor_order_sync" resource="default">
    <column xsi:type="int" name="entity_id"
      unsigned="true" identity="true"/>
    <column xsi:type="int" name="order_id" unsigned="true"/>
    <column xsi:type="varchar" name="status" length="20"/>
    <column xsi:type="int" name="attempts" default="0"/>
    <column xsi:type="text" name="last_error" nullable="true"/>
    <column xsi:type="timestamp" name="updated_at"
      default="CURRENT_TIMESTAMP" on_update="true"/>

    <constraint xsi:type="primary" referenceId="PRIMARY">
      <column name="entity_id"/>
    </constraint>

    <!-- المفتاح: أوردر واحد = صف واحد -->
    <constraint xsi:type="unique" referenceId="VENDOR_ORDER_SYNC_ORDER_ID">
      <column name="order_id"/>
    </constraint>
  </table>
</schema>

<config>
  <topic name="vendor.order.sync"
    request="string"
    handler="Vendor\OrderSync\Model\Consumer::process"/>
</config>

<config>
  <exchange name="magento" type="topic" connection="amqp">
    <binding id="orderSyncBinding"
      topic="vendor.order.sync"
      destinationType="queue"
      destination="vendor.order.sync"/>
  </exchange>
</config>

<config>
  <publisher topic="vendor.order.sync">
    <connection name="amqp" exchange="magento"/>
  </publisher>
</config>

<config>
  <consumer name="vendorOrderSync"
    queue="vendor.order.sync"
    connection="amqp"
    handler="Vendor\OrderSync\Model\Consumer::process"/>
</config>

<config>
  <!-- بعد تأكيد الحفظ، مش قبله -->
  <event name="sales_order_save_commit_after">
    <observer name="vendor_queue_order_sync"
      instance="Vendor\OrderSync\Observer\QueueOrderSync"/>
  </event>
</config>

namespace Vendor\OrderSync\Observer;

use Magento\Framework\Event\Observer;
use Magento\Framework\Event\ObserverInterface;
use Magento\Framework\MessageQueue\PublisherInterface;
use Psr\Log\LoggerInterface;

class QueueOrderSync implements ObserverInterface
{
    private const TOPIC = 'vendor.order.sync';

    public function __construct(
        private readonly PublisherInterface $publisher,
        private readonly LoggerInterface $logger
    ) {}

    public function execute(Observer $observer): void
    {
        $order = $observer->getEvent()->getOrder();

        if (!$order || !$order->getId()) {
            return;
        }

        // الحدث بيتطلق مع أي حفظ — ننشر عند الإنشاء بس
        if (!$order->isObjectNew()) {
            return;
        }

        try {
            $this->publisher->publish(self::TOPIC, json_encode([
                'order_id'  => (int) $order->getId(),
                'increment' => $order->getIncrementId(),
                'store_id'  => (int) $order->getStoreId(),
                'queued_at' => gmdate('c'),
            ]));
        } catch (\Exception $e) {
            // الأوردر اتعمل بنجاح — منوقّعوش بسبب الطابور
            $this->logger->critical(
                'Failed to queue order sync: ' . $e->getMessage(),
                ['order_id' => $order->getId()]
            );
        }
    }
}

namespace Vendor\OrderSync\Model;

class Consumer
{
    public function __construct(
        private readonly OrderRepositoryInterface $orderRepository,
        private readonly SyncLog $syncLog,
        private readonly ErpClient $erp,
        private readonly LoggerInterface $logger
    ) {}

    public function process(string $message): void
    {
        $data = json_decode($message, true);
        $orderId = (int) ($data['order_id'] ?? 0);

        if (!$orderId) {
            // رسالة تالفة — سجّل واخرج، متحاولش تاني
            $this->logger->error('Malformed message', ['raw' => $message]);
            return;
        }

        // 1. الحماية من التكرار
        if ($this->syncLog->isDone($orderId)) {
            return;  // اتعالج قبل كده
        }

        try {
            // 2. حمّل النسخة الحالية
            $order = $this->orderRepository->get($orderId);

            // 3. العملية الخارجية
            $this->erp->sendOrder($order);

            // 4. سجّل النجاح
            $this->syncLog->markDone($orderId);

        } catch (NoSuchEntityException $e) {
            // الأوردر مش موجود — إعادة المحاولة مش هتنفع
            $this->syncLog->markFailed($orderId, 'order not found');

        } catch (\Exception $e) {
            // فشل مؤقت — سجّله عشان الـ cron يعيد
            $this->syncLog->markRetry($orderId, $e->getMessage());
            $this->logger->error('ERP sync failed', [
                'order_id' => $orderId,
                'error'    => $e->getMessage(),
            ]);
        }
    }
}

public function execute(): void
{
    // اللي فشل ولسه محاولاته أقل من الحد
    $rows = $this->syncLog->getRetryable(maxAttempts: 5);

    foreach ($rows as $row) {
        // تباعد متزايد: 2^n دقيقة
        $waitMinutes = 2 ** (int) $row['attempts'];
        if (!$this->waitedEnough($row, $waitMinutes)) {
            continue;
        }

        $this->publisher->publish('vendor.order.sync', json_encode([
            'order_id' => (int) $row['order_id'],
            'retry'    => true,
        ]));
    }
}

# شغّال باستمرار
bin/magento queue:consumers:start vendorOrderSync

# عدد محدود — مفيد للاختبار
bin/magento queue:consumers:start vendorOrderSync \
  --max-messages=10

# شوف كل الـ consumers المتاحة
bin/magento queue:consumers:list

'cron_consumers_runner' => [
    'cron_run'     => true,
    'max_messages' => 1000,
    'consumers'    => ['vendorOrderSync'],
],

# طول الطوابير (من سيرفر RabbitMQ)
rabbitmqctl list_queues name messages consumers

# هل الـ consumer شغّال؟
ps aux | grep queue:consumers:start

# أخطاء المزامنة في الـ log
grep "ERP sync failed" var/log/*.log | tail -20

-- توزيع الحالات
SELECT status, COUNT(*) FROM vendor_order_sync
GROUP BY status;

-- الأوردرات اللي فشلت نهائياً
SELECT order_id, attempts, last_error, updated_at
FROM vendor_order_sync
WHERE status = 'failed'
ORDER BY updated_at DESC;

-- أوردرات مالهاش صف أصلاً (رسالة ضاعت)
SELECT o.entity_id FROM sales_order o
LEFT JOIN vendor_order_sync s ON s.order_id = o.entity_id
WHERE s.order_id IS NULL
  AND o.created_at < NOW() - INTERVAL 1 HOUR;

# 1. اعمل أوردر من الواجهة أو برمجياً

# 2. شوف الرسالة وصلت الطابور
rabbitmqctl list_queues name messages

# 3. شغّل رسالة واحدة وراقب
bin/magento queue:consumers:start vendorOrderSync \
  --max-messages=1

# 4. اتأكد من التسجيل
#    SELECT * FROM vendor_order_sync ORDER BY entity_id DESC LIMIT 1;

# 5. اختبر الحماية من التكرار: أعد نشر نفس الرسالة
#    المفروض الـ consumer يتجاهلها