Skip to content
advanced Phase 118 · Advanced Projects

Project - ERP Integration

Build an ERP integration with bidirectional data sync, message queues, conflict resolution, and error handling

2h
0 problems
Topic Progress 0%

Integration Architecture

Architecture Overview

┌────────────────────────────────────────────────────────────┐
│                      Magento 2 Store                       │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐  │
│  │ Product  │  │ Customer │  │  Order   │  │Inventory │  │
│  │ Module   │  │ Module   │  │ Module   │  │  Module  │  │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘  │
│       │              │              │              │        │
│       └──────────────┼──────────────┼──────────────┘        │
│                      │              │                       │
│              ┌───────┴──────────────┴───────┐               │
│              │    Integration Hub (MQ)       │               │
│              └───────────────┬───────────────┘               │
└──────────────────────────────┼─────────────────────────────┘
                               │
                        ┌──────┴──────┐
                        │  Message    │
                        │  Queue      │
                        │  (RabbitMQ) │
                        └──────┬──────┘
                               │
                        ┌──────┴──────┐
                        │  ERP Sync   │
                        │  Service    │
                        └──────┬──────┘
                               │
              ┌────────────────┼────────────────┐
              │                │                │
       ┌──────┴──────┐ ┌──────┴──────┐ ┌──────┴──────┐
       │    SAP      │ │   Oracle    │ │  Custom     │
       │  Adapter    │ │  Adapter    │ │  ERP        │
       └─────────────┘ └─────────────┘ └─────────────┘

Module Structure

app/code/Vendor/ErpIntegration/
├── registration.php
├── etc/
│   ├── module.xml
│   ├── di.xml
│   ├── communication.xml
│   └── queue_consumer.xml
├── Api/
│   ├── ErpAdapterInterface.php
│   ├── SyncServiceInterface.php
│   └── Data/
│       ├── SyncResultInterface.php
│       └── SyncStateInterface.php
├── Model/
│   ├── Erp/
│   │   ├── SapAdapter.php
│   │   ├── OracleAdapter.php
│   │   └── ErpAdapterFactory.php
│   ├── Sync/
│   │   ├── ProductSync.php
│   │   ├── OrderSync.php
│   │   ├── CustomerSync.php
│   │   └── InventorySync.php
│   ├── Conflict/
│   │   ├── ConflictResolver.php
│   │   └── ConflictLog.php
│   └── Resilience/
│       ├── CircuitBreaker.php
│       └── RetryHandler.php
├── Queue/
│   ├── Publisher/
│   │   ├── ProductPublisher.php
│   │   └── OrderPublisher.php
│   └── Consumer/
│       ├── ErpSyncConsumer.php
│       └── OrderPushConsumer.php
└── Observer/
    └── DataChangeObserver.php

ERP Adapter Interface

<?php
namespace Vendor\ErpIntegration\Api;

use Vendor\ErpIntegration\Api\Data\SyncResultInterface;

interface ErpAdapterInterface
{
    public function getAdapterCode(): string;

    public function pullProduct(string $sku): ?array;
    public function pushProduct(array $productData): bool;

    public function pullInventory(string $sku): ?array;
    public function pushInventory(array $inventoryData): bool;

    public function pullCustomers(int $page, int $pageSize): array;
    public function pushOrder(array $orderData): SyncResultInterface;

    public function pullOrders(int $page, int $pageSize): array;
    public function pullOrderStatus(string $incrementId): ?array;
}

Message Queue Integration

Queue Configuration

<!-- etc/communication.xml -->
<?xml version="1.0"?>
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/communication.xsd">
    <topic name="erp.sync.product.push" sync="false">
        <handler name="default" type="Vendor\ErpIntegration\Queue\Publisher\ProductPublisher" method="publish"/>
    </topic>
    <topic name="erp.sync.product.pull" sync="false">
        <handler name="default" type="Vendor\ErpIntegration\Queue\Publisher\ProductPublisher" method="publishPull"/>
    </topic>
    <topic name="erp.sync.order.push" sync="false">
        <handler name="default" type="Vendor\ErpIntegration\Queue\Publisher\OrderPublisher" method="publish"/>
    </topic>
    <topic name="erp.sync.inventory.update" sync="false">
        <handler name="default" type="Vendor\ErpIntegration\Queue\Publisher\ProductPublisher" method="publishInventory"/>
    </topic>
</config>

<!-- etc/queue_consumer.xml -->
<?xml version="1.0"?>
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/consumer.xsd">
    <consumer name="erp.sync.product" queue="erp.sync.product.push" handler="Vendor\ErpIntegration\Queue\Consumer\ErpSyncConsumer::processProduct" connection="amqp" maxMessages="100"/>
    <consumer name="erp.sync.order" queue="erp.sync.order.push" handler="Vendor\ErpIntegration\Queue\Consumer\OrderPushConsumer::process" connection="amqp" maxMessages="50"/>
    <consumer name="erp.sync.inventory" queue="erp.sync.inventory.update" handler="Vendor\ErpIntegration\Queue\Consumer\ErpSyncConsumer::processInventory" connection="amqp" maxMessages="200"/>
</config>

Publisher

<?php
namespace Vendor\ErpIntegration\Queue\Publisher;

use Magento\Framework\MessageQueue\PublisherInterface;

class ProductPublisher
{
    public function __construct(
        private PublisherInterface $publisher,
    ) {
    }

    public function publish(array $productData): void
    {
        $this->publisher->publish(
            'erp.sync.product.push',
            json_encode([
                'action' => 'push',
                'sku' => $productData['sku'],
                'data' => $productData,
                'timestamp' => time(),
            ])
        );
    }

    public function publishPull(string $sku): void
    {
        $this->publisher->publish(
            'erp.sync.product.pull',
            json_encode([
                'action' => 'pull',
                'sku' => $sku,
                'timestamp' => time(),
            ])
        );
    }

    public function publishInventory(array $inventoryData): void
    {
        $this->publisher->publish(
            'erp.sync.inventory.update',
            json_encode([
                'action' => 'inventory_update',
                'items' => $inventoryData,
                'timestamp' => time(),
            ])
        );
    }
}

Consumer

<?php
namespace Vendor\ErpIntegration\Queue\Consumer;

use Vendor\ErpIntegration\Api\SyncServiceInterface;
use Vendor\ErpIntegration\Model\Resilience\CircuitBreaker;
use Vendor\ErpIntegration\Model\Resilience\RetryHandler;
use Psr\Log\LoggerInterface;

class ErpSyncConsumer
{
    public function __construct(
        private SyncServiceInterface $syncService,
        private CircuitBreaker $circuitBreaker,
        private RetryHandler $retryHandler,
        private LoggerInterface $logger,
    ) {
    }

    public function processProduct(string $messageBody): void
    {
        $message = json_decode($messageBody, true);

        $this->logger->info('ERP Sync: Processing product', [
            'sku' => $message['sku'] ?? 'unknown',
            'action' => $message['action'] ?? 'unknown',
        ]);

        $this->circuitBreaker->call(
            function () use ($message) {
                return $this->retryHandler->executeWithRetry(
                    function () use ($message) {
                        if ($message['action'] === 'push') {
                            $this->syncService->pushProductToErp($message['sku'], $message['data']);
                        } elseif ($message['action'] === 'pull') {
                            $this->syncService->pullProductFromErp($message['sku']);
                        }
                    }
                );
            },
            function () use ($message) {
                $this->logger->critical('ERP Sync: Failed after retries', [
                    'sku' => $message['sku'],
                    'message' => $messageBody,
                ]);
            }
        );
    }

    public function processInventory(string $messageBody): void
    {
        $message = json_decode($messageBody, true);

        foreach ($message['items'] as $item) {
            try {
                $this->syncService->syncInventory($item['sku'], $item['quantity']);
            } catch (\Exception $e) {
                $this->logger->error('ERP Sync: Inventory sync failed', [
                    'sku' => $item['sku'],
                    'error' => $e->getMessage(),
                ]);
            }
        }
    }
}

Conflict Resolution and Data Mapping

Conflict Resolution

<?php
namespace Vendor\ErpIntegration\Model\Conflict;

interface ConflictResolverInterface
{
    public function resolve(array $localData, array $remoteData, array $metadata): array;
}

class LastModifiedResolver implements ConflictResolverInterface
{
    public function resolve(array $localData, array $remoteData, array $metadata): array
    {
        $localModified = strtotime($metadata['local_modified_at'] ?? 'now');
        $remoteModified = strtotime($metadata['remote_modified_at'] ?? 'now');

        if ($remoteModified > $localModified) {
            // Remote is newer - use remote data
            return [
                'strategy' => 'remote_wins',
                'merged_data' => $remoteData,
                'conflict' => false,
            ];
        }

        if ($localModified > $remoteModified) {
            // Local is newer - use local data
            return [
                'strategy' => 'local_wins',
                'merged_data' => $localData,
                'conflict' => false,
            ];
        }

        // Same timestamp - merge with field-level comparison
        return $this->mergeByField($localData, $remoteData, $metadata);
    }

    private function mergeByField(array $local, array $remote, array $metadata): array
    {
        $merged = $local;
        $conflicts = [];

        foreach ($remote as $field => $value) {
            if (isset($local[$field]) && $local[$field] !== $value) {
                $conflicts[] = $field;
                // Use last modified per field if available
                $fieldModifiedAt = $metadata['field_modified'][$field] ?? null;
                if ($fieldModifiedAt) {
                    $merged[$field] = $value;
                }
            } else {
                $merged[$field] = $value;
            }
        }

        return [
            'strategy' => 'field_merge',
            'merged_data' => $merged,
            'conflict' => !empty($conflicts),
            'conflicted_fields' => $conflicts,
        ];
    }
}

// Conflict Logger
class ConflictLog
{
    public function __construct(
        private \Magento\Framework\App\ResourceConnection $resource,
    ) {
    }

    public function log(string $entityType, string $sku, array $localData, array $remoteData, string $resolution): void
    {
        $connection = $this->resource->getConnection();
        $tableName = $this->resource->getTableName('erp_sync_conflict_log');

        $connection->insert($tableName, [
            'entity_type' => $entityType,
            'sku' => $sku,
            'local_data' => json_encode($localData),
            'remote_data' => json_encode($remoteData),
            'resolution' => $resolution,
            'created_at' => (new \DateTime())->format('Y-m-d H:i:s'),
        ]);
    }
}

Data Mapping Service

<?php
namespace Vendor\ErpIntegration\Model\Mapping;

class ProductDataMapper
{
    private array $fieldMap = [
        'sku' => 'MATNR',
        'name' => 'MAKTX',
        'price' => 'KBETR',
        'weight' => 'NTGEW',
        'description' => 'LONG_TEXT',
        'status' => 'MASTA',
    ];

    public function mapToErp(array $magentoData): array
    {
        $erpData = [];
        foreach ($this->fieldMap as $magentoField => $erpField) {
            if (isset($magentoData[$magentoField])) {
                $erpData[$erpField] = $this->transformToErp($magentoField, $magentoData[$magentoField]);
            }
        }
        return $erpData;
    }

    public function mapToMagento(array $erpData): array
    {
        $magentoData = [];
        foreach ($this->fieldMap as $magentoField => $erpField) {
            if (isset($erpData[$erpField])) {
                $magentoData[$magentoField] = $this->transformToMagento($magentoField, $erpData[$erpField]);
            }
        }
        return $magentoData;
    }

    private function transformToErp(string $field, $value)
    {
        return match($field) {
            'price' => number_format((float)$value, 2, '.', ''),
            'status' => $value === '1' ? 'A' : 'I',
            default => (string)$value,
        };
    }

    private function transformToMagento(string $field, $value)
    {
        return match($field) {
            'price' => (float)$value,
            'status' => $value === 'A' ? '1' : '0',
            default => $value,
        };
    }
}

Resilience Patterns and Monitoring

Circuit Breaker

<?php
namespace Vendor\ErpIntegration\Model\Resilience;

class CircuitBreaker
{
    private int $failureThreshold;
    private int $recoveryTimeout;
    private int $failureCount = 0;
    private string $state = 'closed';
    private ?\DateTime $lastFailureTime = null;

    public function __construct(
        private \Magento\Framework\App\Config\ScopeConfigInterface $scopeConfig,
    ) {
        $this->failureThreshold = (int) $this->scopeConfig->getValue('erp_integration/circuit_breaker/threshold') ?: 5;
        $this->recoveryTimeout = (int) $this->scopeConfig->getValue('erp_integration/circuit_breaker/timeout') ?: 60;
    }

    public function call(callable $operation, callable $fallback)
    {
        if ($this->state === 'open') {
            if ($this->shouldAttemptRecovery()) {
                $this->state = 'half-open';
            } else {
                return $fallback();
            }
        }

        try {
            $result = $operation();
            $this->onSuccess();
            return $result;
        } catch (\Exception $e) {
            $this->onFailure();
            return $fallback();
        }
    }

    private function onSuccess(): void
    {
        $this->failureCount = 0;
        $this->state = 'closed';
    }

    private function onFailure(): void
    {
        $this->failureCount++;
        $this->lastFailureTime = new \DateTime();

        if ($this->failureCount >= $this->failureThreshold) {
            $this->state = 'open';
        }
    }

    private function shouldAttemptRecovery(): bool
    {
        if (!$this->lastFailureTime) {
            return false;
        }

        $elapsed = (new \DateTime())->getTimestamp() - $this->lastFailureTime->getTimestamp();
        return $elapsed >= $this->recoveryTimeout;
    }

    public function getState(): string
    {
        return $this->state;
    }
}

Retry Handler

<?php
namespace Vendor\ErpIntegration\Model\Resilience;

class RetryHandler
{
    private int $maxRetries;
    private int $baseDelay;
    private int $maxDelay;

    public function __construct(
        private \Magento\Framework\App\Config\ScopeConfigInterface $scopeConfig,
    ) {
        $this->maxRetries = (int) $this->scopeConfig->getValue('erp_integration/retry/max_retries') ?: 3;
        $this->baseDelay = (int) $this->scopeConfig->getValue('erp_integration/retry/base_delay') ?: 1000;
        $this->maxDelay = (int) $this->scopeConfig->getValue('erp_integration/retry/max_delay') ?: 30000;
    }

    public function executeWithRetry(callable $operation): mixed
    {
        $attempt = 0;
        $lastException = null;

        while ($attempt <= $this->maxRetries) {
            try {
                return $operation();
            } catch (\Exception $e) {
                $lastException = $e;
                $attempt++;

                if ($attempt <= $this->maxRetries) {
                    $delay = min(
                        $this->baseDelay * pow(2, $attempt - 1),
                        $this->maxDelay
                    );
                    usleep($delay * 1000);
                }
            }
        }

        throw $lastException;
    }
}

Monitoring Dashboard Data

<?php
namespace Vendor\ErpIntegration\Model\Monitor;

class SyncMetrics
{
    public function __construct(
        private \Magento\Framework\App\ResourceConnection $resource,
    ) {
    }

    public function getSyncStats(): array
    {
        $connection = $this->resource->getConnection();
        $tableName = $this->resource->getTableName('erp_sync_log');

        $totalSynced = $connection->fetchOne(
            "SELECT COUNT(*) FROM {$tableName} WHERE status = 'success'"
        );

        $failedSyncs = $connection->fetchOne(
            "SELECT COUNT(*) FROM {$tableName} WHERE status = 'failed'"
        );

        $pendingSyncs = $connection->fetchOne(
            "SELECT COUNT(*) FROM {$tableName} WHERE status = 'pending'"
        );

        $recentErrors = $connection->fetchAll(
            "SELECT * FROM {$tableName} WHERE status = 'failed' ORDER BY created_at DESC LIMIT 10"
        );

        return [
            'total_synced' => $totalSynced,
            'failed' => $failedSyncs,
            'pending' => $pendingSyncs,
            'success_rate' => $totalSynced > 0
                ? round(($totalSynced / ($totalSynced + $failedSyncs)) * 100, 2)
                : 0,
            'recent_errors' => $recentErrors,
        ];
    }
}

Quiz

1. What pattern prevents cascading ERP failures?

Question 1 options

2. What handles concurrent data changes?

Question 2 options

3. What is exponential backoff?

Question 3 options

Flashcards

Question

What is bidirectional sync?

Click to reveal answer

Answer

Data flows both Magento→ERP and ERP→Magento

Question

What is circuit breaker?

Click to reveal answer

Answer

Pattern that stops calls to failing service after threshold

Question

What is conflict resolution?

Click to reveal answer

Answer

Strategy to merge local and remote data changes

Question

What is message queue async?

Click to reveal answer

Answer

Processing sync tasks in background via RabbitMQ

Question

What is data mapping?

Click to reveal answer

Answer

Translating field names/formats between systems

Revision Notes

Key Takeaways

  • 1. Bidirectional sync requires conflict resolution strategies
  • 2. Message queues enable async processing and fault tolerance
  • 3. Circuit breaker and retry patterns improve resilience
  • 4. Data mapping translates field names/formats between systems
  • 5. Monitoring dashboards track sync success rates and errors

Interview Tips

  • • Explain bidirectional sync architecture and conflict resolution
  • • Describe circuit breaker state machine (closed → open → half-open)
  • • Discuss message queue benefits for ERP integration
  • • Talk about data mapping challenges between different systems

Cheat Sheet

ERP Integration:
  Adapter Pattern → SAP/Oracle/Custom
  Message Queue → Async sync processing
  Circuit Breaker → Prevent cascading failures
  Retry Handler → Exponential backoff

Conflict Resolution:
  Last-Modified → Remote wins if newer
  Field-Level → Merge per field
  Manual → Queue for review

Sync Architecture:
  Publisher → MQ → Consumer → ERP
  Bidirectional: Push + Pull
  Monitoring: Success rate, error log

Data Mapping:
  Field map: Magento ↔ ERP
  Transform: Type conversion
  Validate: Business rules