Fix inventory update with new API endpoints

composer.json

@@ -19,7 +19,7 @@
         "psr/log": "^1.0 || ^2.0 || ^3.0",
         "setono/composite-compiler-pass": "^1.1",
         "setono/doctrine-orm-trait": "^1.1",
-        "setono/peak-wms-php-sdk": "^1.1",
+        "setono/peak-wms-php-sdk": "^1.2",
         "sylius/admin-bundle": "^1.0",
         "sylius/core": "^1.0",
         "sylius/core-bundle": "^1.0",

src/EventSubscriber/Workflow/InventoryUpdate/CompleteSubscriber.php

@@ -26,9 +26,5 @@ public function set(CompletedEvent $event): void
         Assert::isInstanceOf($inventoryUpdate, InventoryUpdateInterface::class);
 
         $inventoryUpdate->setCompletedAt(new \DateTimeImmutable());
-
-        if (!$inventoryUpdate->hasErrors()) {
-            $inventoryUpdate->setNextUpdateThreshold($inventoryUpdate->getProcessingStartedAt());
-        }
     }
 }

src/Model/InventoryUpdate.php

@@ -4,7 +4,7 @@
 
 namespace Setono\SyliusPeakPlugin\Model;
 
 class InventoryUpdate implements InventoryUpdateInterface
 {
     protected ?int $id = null;
 
@@ -16,8 +16,6 @@
 
     protected ?\DateTimeInterface $completedAt = null;
 
-    protected ?\DateTimeInterface $nextUpdateThreshold = null;
-
     protected int $productsProcessed = 0;
 
     /** @var list<string>|null */
@@ -71,16 +69,6 @@
         $this->completedAt = $completedAt;
     }
 
-    public function getNextUpdateThreshold(): ?\DateTimeInterface
-    {
-        return $this->nextUpdateThreshold;
-    }
-
-    public function setNextUpdateThreshold(?\DateTimeInterface $nextUpdateThreshold): void
-    {
-        $this->nextUpdateThreshold = $nextUpdateThreshold;
-    }
-
     public function getProductsProcessed(): int
     {
         return $this->productsProcessed;

src/Model/InventoryUpdateInterface.php

@@ -7,7 +7,7 @@
 use Sylius\Component\Resource\Model\ResourceInterface;
 use Sylius\Component\Resource\Model\VersionedInterface;
 
 interface InventoryUpdateInterface extends ResourceInterface, VersionedInterface
 {
     final public const STATE_PENDING = 'pending';
 
@@ -37,13 +37,6 @@
 
     public function setCompletedAt(?\DateTimeInterface $completedAt): void;
 
-    /**
-     * This is the threshold to use when fetching updated products from Peak WMS. If null, it means that all products should be fetched.
-     */
-    public function getNextUpdateThreshold(): ?\DateTimeInterface;
-
-    public function setNextUpdateThreshold(?\DateTimeInterface $nextUpdateThreshold): void;
-
     public function getProductsProcessed(): int;
 
     public function setProductsProcessed(int $productsProcessed): void;

src/Resources/config/doctrine/model/InventoryUpdate.orm.xml

@@ -14,7 +14,6 @@
         <field name="state" type="string"/>
         <field name="processingStartedAt" type="datetime" nullable="true"/>
         <field name="completedAt" type="datetime" nullable="true"/>
-        <field name="nextUpdateThreshold" type="datetime" nullable="true"/>
         <field name="productsProcessed" type="integer"/>
         <field name="warnings" type="json" nullable="true"/>
         <field name="errors" type="json" nullable="true"/>

src/Updater/InventoryUpdater.php

@@ -7,8 +7,8 @@
 use Doctrine\Persistence\ManagerRegistry;
 use Setono\Doctrine\ORMTrait;
 use Setono\PeakWMS\Client\ClientInterface;
-use Setono\PeakWMS\DataTransferObject\Product\Product;
-use Setono\PeakWMS\Request\Query\Product\PageQuery;
+use Setono\PeakWMS\DataTransferObject\Stock\Stock;
+use Setono\PeakWMS\Request\Query\KeySetPageQuery;
 use Setono\SyliusPeakPlugin\Provider\InventoryUpdateProviderInterface;
 use Setono\SyliusPeakPlugin\Workflow\InventoryUpdateWorkflow;
 use Sylius\Component\Core\Model\ProductVariant;
@@ -43,20 +43,11 @@ public function update(ProductVariantInterface $productVariant): void
 
         $collection = $this
             ->client
-            ->product()
-            ->getByProductId($productCode)
-            ->filter(fn (Product $product) => $product->variantId === $variantCode)
+            ->stock()
+            ->getByProductId($productCode, $variantCode)
         ;
 
-        if (count($collection) !== 1) {
-            throw new \RuntimeException(sprintf(
-                'The product with id %s either does not have a variant with id/code %s or has multiple products with the same variant id/code',
-                $productCode,
-                $variantCode,
-            ));
-        }
-
-        $this->map($collection[0], $productVariant);
+        $this->updateOnHand((int) $collection->sum(fn (Stock $stock): int => (int) $stock->quantity), $productVariant);
 
         $this->getManager($productVariant)->flush();
     }
@@ -75,8 +66,8 @@ public function updateAll(bool $onlyUpdated = true): void
             $productVariantRepository = $this->getRepository(ProductVariant::class);
 
             $i = 0;
-            $products = $this->client->product()->iterate(PageQuery::create(updatedAfter: $inventoryUpdate->getNextUpdateThreshold()));
-            foreach ($products as $product) {
+            $stock = $this->client->stock()->iterate(KeySetPageQuery::create());
+            foreach ($stock as $item) {
                 ++$i;
 
                 if ($i % 100 === 0) {
@@ -90,27 +81,27 @@ public function updateAll(bool $onlyUpdated = true): void
                 }
 
                 try {
-                    Assert::notNull($product->variantId, sprintf(
-                        'Product with id %d does not have a variant id. It is expected that Peak WMS has the same structure of products as Sylius, namely that all products at least have one variant.',
-                        (int) $product->id,
+                    Assert::notNull($item->variantId, sprintf(
+                        'Stock with id %d does not have a variant id.',
+                        (int) $item->id,
                     ));
 
-                    $productVariant = $productVariantRepository->findOneBy(['code' => $product->variantId]);
+                    $productVariant = $productVariantRepository->findOneBy(['code' => $item->variantId]);
                     Assert::notNull(
                         $productVariant,
-                        sprintf('Product variant with code %s does not exist', $product->variantId),
+                        sprintf('Product variant with code %s does not exist', $item->variantId),
                     );
 
-                    if ($product->orderedByCustomers !== $productVariant->getOnHold()) {
+                    if ($item->reservedQuantity !== $productVariant->getOnHold()) {
                         $inventoryUpdate->addWarning(sprintf(
                             'Product variant with code %s has %d on hold in Sylius and %d on hold in Peak WMS',
-                            $product->variantId,
+                            $item->variantId,
                             (int) $productVariant->getOnHold(),
-                            (int) $product->orderedByCustomers,
+                            (int) $item->reservedQuantity,
                         ));
                     }
 
-                    $this->map($product, $productVariant);
+                    $this->updateOnHand((int) $item->quantity, $productVariant);
                 } catch (\Throwable $e) {
                     $inventoryUpdate->addError($e->getMessage());
                 }
@@ -125,6 +116,7 @@ public function updateAll(bool $onlyUpdated = true): void
             $this->inventoryUpdateTransition(InventoryUpdateWorkflow::TRANSITION_FAIL);
         } finally {
             $manager->flush();
+            $manager->clear();
         }
     }
 
@@ -136,8 +128,8 @@ private function inventoryUpdateTransition(string $transition): void
         $this->getManager($inventoryUpdate)->flush();
     }
 
-    private function map(Product $product, ProductVariantInterface $productVariant): void
+    private function updateOnHand(int $quantity, ProductVariantInterface $productVariant): void
     {
-        $productVariant->setOnHand((int) $product->availableToSell + (int) $productVariant->getOnHold());
+        $productVariant->setOnHand($quantity + (int) $productVariant->getOnHold());
     }
 }