esun3d/app/code/Esun/LingxingErp/Model/Inventory/InventorySynchronizer.php

<?php
declare(strict_types=1);

namespace Esun\LingxingErp\Model\Inventory;

use Esun\LingxingErp\Model\Sku\ProductErpSkuRepository;
use Esun\LingxingErp\Model\Sync\LogWriter;
use Esun\LingxingErp\Model\Warehouse\WarehouseRepository;
use Magento\Framework\Lock\LockManagerInterface;
use Magento\InventoryApi\Api\Data\SourceItemInterface;
use Magento\InventoryApi\Api\Data\SourceItemInterfaceFactory;
use Magento\InventoryApi\Api\SourceItemsSaveInterface;

class InventorySynchronizer
{
    private const LOCK_NAME = 'esun_lingxing_inventory_sync';
    private const SAVE_BATCH_SIZE = 500;

    public function __construct(
        private readonly WarehouseRepository $warehouseRepository,
        private readonly InventoryFetcher $fetcher,
        private readonly ProductErpSkuRepository $productErpSkuRepository,
        private readonly SourceItemInterfaceFactory $sourceItemFactory,
        private readonly SourceItemsSaveInterface $sourceItemsSave,
        private readonly LogWriter $logWriter,
        private readonly LockManagerInterface $lockManager
    ) {
    }

    /** @return array{warehouses:int,items:int,unmapped:int,errors:array<int, string>} */
    public function synchronizeAll(): array
    {
        if (!$this->lockManager->lock(self::LOCK_NAME, 0)) {
            throw new \RuntimeException('Inventory synchronization is already running.');
        }
        $summary = ['warehouses' => 0, 'items' => 0, 'unmapped' => 0, 'errors' => []];
        try {
            foreach ($this->warehouseRepository->getInventoryTargets() as $warehouse) {
                $wid = (int)$warehouse['wid'];
                $started = microtime(true);
                try {
                    $result = $this->synchronizeWarehouse($warehouse);
                    $summary['warehouses']++;
                    $summary['items'] += $result['items'];
                    $summary['unmapped'] += $result['unmapped'];
                    $this->warehouseRepository->markInventorySynchronized($wid);
                    $this->logWriter->write([
                        'direction' => 'pull',
                        'operation' => 'inventory_sync',
                        'entity_type' => 'inventory',
                        'external_id' => (string)$wid,
                        'status' => 'success',
                        'attempt_number' => 1,
                        'request_payload' => [
                            'wid' => $wid,
                            'source_code' => (string)$warehouse['source_code'],
                            'quantity_field' => 'product_valid_num',
                        ],
                        'response_payload' => $result,
                        'duration_ms' => (int)round((microtime(true) - $started) * 1000),
                    ]);
                } catch (\Throwable $exception) {
                    $summary['errors'][$wid] = $exception->getMessage();
                    $this->logWriter->write([
                        'direction' => 'pull',
                        'operation' => 'inventory_sync',
                        'entity_type' => 'inventory',
                        'external_id' => (string)$wid,
                        'status' => 'failed',
                        'attempt_number' => 1,
                        'error_code' => (string)$exception->getCode(),
                        'error_message' => $exception->getMessage(),
                        'request_payload' => [
                            'wid' => $wid,
                            'source_code' => (string)$warehouse['source_code'],
                            'quantity_field' => 'product_valid_num',
                        ],
                        'duration_ms' => (int)round((microtime(true) - $started) * 1000),
                    ]);
                }
            }
        } finally {
            $this->lockManager->unlock(self::LOCK_NAME);
        }
        return $summary;
    }

    /**
     * @param array<string, mixed> $warehouse
     * @return array{items:int,unmapped:int,zeroed:int,preserved_without_erp_sku:int,unmapped_erp_skus:string[]}
     */
    private function synchronizeWarehouse(array $warehouse): array
    {
        $wid = (int)$warehouse['wid'];
        $sourceCode = (string)$warehouse['source_code'];
        $websiteIds = $this->warehouseRepository->getWebsiteIdsForSource($sourceCode);
        if ($websiteIds === []) {
            throw new \RuntimeException(sprintf(
                'Magento Source "%s" is not assigned to a Stock with a Website sales channel.',
                $sourceCode
            ));
        }

        $rows = $this->fetcher->fetchAll($wid);
        $quantitiesBySku = [];
        $unmapped = [];
        foreach ($rows as $row) {
            $erpSku = trim((string)($row['sku'] ?? ''));
            if ($erpSku === '') {
                continue;
            }
            $magentoSkus = $this->resolveMagentoSkus($websiteIds, $erpSku);
            if ($magentoSkus === []) {
                $unmapped[$erpSku] = true;
                continue;
            }
            $quantity = max(0.0, (float)($row['product_valid_num'] ?? 0));
            foreach ($magentoSkus as $magentoSku) {
                // The customer accepts that every Magento SKU paired to the ERP SKU receives the full ERP quantity.
                $quantitiesBySku[$magentoSku] = $quantity;
            }
        }

        $existingSkus = $this->warehouseRepository->getExistingSourceItemSkus($sourceCode);
        $mappedSkus = $this->getMappedMagentoSkus($websiteIds);
        $zeroCandidates = array_values(array_intersect($existingSkus, $mappedSkus));
        $preservedWithoutErpSku = array_values(array_diff($existingSkus, $mappedSkus));
        $zeroed = array_values(array_diff($zeroCandidates, array_keys($quantitiesBySku)));
        foreach ($zeroed as $sku) {
            $quantitiesBySku[$sku] = 0.0;
        }
        $sourceItems = [];
        foreach ($quantitiesBySku as $sku => $quantity) {
            $sourceItem = $this->sourceItemFactory->create();
            $sourceItem->setSourceCode($sourceCode);
            $sourceItem->setSku($sku);
            $sourceItem->setQuantity($quantity);
            $sourceItem->setStatus($quantity > 0
                ? SourceItemInterface::STATUS_IN_STOCK
                : SourceItemInterface::STATUS_OUT_OF_STOCK);
            $sourceItems[] = $sourceItem;
        }
        foreach (array_chunk($sourceItems, self::SAVE_BATCH_SIZE) as $batch) {
            $this->sourceItemsSave->execute($batch);
        }
        return [
            'items' => count($quantitiesBySku),
            'unmapped' => count($unmapped),
            'zeroed' => count($zeroed),
            'preserved_without_erp_sku' => count($preservedWithoutErpSku),
            'unmapped_erp_skus' => array_slice(array_keys($unmapped), 0, 200),
        ];
    }

    /** @param int[] $websiteIds */
    private function resolveMagentoSkus(array $websiteIds, string $erpSku): array
    {
        $matches = [];
        foreach ($websiteIds as $websiteId) {
            foreach ($this->productErpSkuRepository->findMskusByErpSku($websiteId, $erpSku) as $sku) {
                $matches[$sku] = true;
            }
        }
        return array_map('strval', array_keys($matches));
    }

    /** @param int[] $websiteIds @return string[] */
    private function getMappedMagentoSkus(array $websiteIds): array
    {
        $mapped = [];
        foreach ($websiteIds as $websiteId) {
            foreach ($this->productErpSkuRepository->getWebsiteProductSkusWithErpSku($websiteId) as $sku) {
                $mapped[$sku] = true;
            }
        }
        return array_map('strval', array_keys($mapped));
    }
}