From d5e99b4f53cf7896dd37d0ea60c0ae41479d8958 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Joachim=20L=C3=B8vgaard?= <joachim@loevgaard.dk>
Date: Thu, 27 Jun 2024 13:34:06 +0200
Subject: [PATCH] Create more inventory handling

---
 src/Command/UpdateInventoryCommand.php        |  32 ++++
 src/DependencyInjection/Configuration.php     |  15 ++
 .../SetonoSyliusPeakExtension.php             |   3 +-
 .../InventoryUpdate/CompleteSubscriber.php    |  34 +++++
 .../InventoryUpdate/ProcessSubscriber.php     |  30 ++++
 .../InventoryUpdate/ResetSubscriber.php       |  33 ++++
 src/Message/Command/UpdateInventory.php       |  21 ++-
 .../CommandHandler/UpdateInventoryHandler.php |  43 ++----
 src/Model/InventoryUpdate.php                 | 141 ++++++++++++++++++
 src/Model/InventoryUpdateInterface.php        |  76 ++++++++++
 src/Provider/InventoryUpdateProvider.php      |  43 ++++++
 .../InventoryUpdateProviderInterface.php      |  15 ++
 .../doctrine/model/InventoryUpdate.orm.xml    |  22 +++
 src/Resources/config/services.xml             |   1 +
 src/Resources/config/services/client.xml      |   2 +
 src/Resources/config/services/command.xml     |  12 +-
 .../config/services/event_subscriber.xml      |  13 ++
 src/Resources/config/services/message.xml     |   7 +
 src/Resources/config/services/provider.xml    |   8 +
 src/Resources/config/services/updater.xml     |  15 ++
 src/Updater/InventoryUpdater.php              | 129 ++++++++++++++++
 src/Updater/InventoryUpdaterInterface.php     |  22 +++
 .../StockAdjustmentWebhookHandler.php         |   2 +-
 src/Workflow/InventoryUpdateWorkflow.php      |  86 +++++++++++
 24 files changed, 759 insertions(+), 46 deletions(-)
 create mode 100644 src/Command/UpdateInventoryCommand.php
 create mode 100644 src/EventSubscriber/Workflow/InventoryUpdate/CompleteSubscriber.php
 create mode 100644 src/EventSubscriber/Workflow/InventoryUpdate/ProcessSubscriber.php
 create mode 100644 src/EventSubscriber/Workflow/InventoryUpdate/ResetSubscriber.php
 create mode 100644 src/Model/InventoryUpdate.php
 create mode 100644 src/Model/InventoryUpdateInterface.php
 create mode 100644 src/Provider/InventoryUpdateProvider.php
 create mode 100644 src/Provider/InventoryUpdateProviderInterface.php
 create mode 100644 src/Resources/config/doctrine/model/InventoryUpdate.orm.xml
 create mode 100644 src/Resources/config/services/updater.xml
 create mode 100644 src/Updater/InventoryUpdater.php
 create mode 100644 src/Updater/InventoryUpdaterInterface.php
 create mode 100644 src/Workflow/InventoryUpdateWorkflow.php

diff --git a/src/Command/UpdateInventoryCommand.php b/src/Command/UpdateInventoryCommand.php
new file mode 100644
index 0000000..d46b744
--- /dev/null
+++ b/src/Command/UpdateInventoryCommand.php
@@ -0,0 +1,32 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Setono\SyliusPeakPlugin\Command;
+
+use Setono\SyliusPeakPlugin\Updater\InventoryUpdaterInterface;
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+#[AsCommand(
+    name: 'setono:sylius-peak-wms:update-inventory',
+    description: 'This will update the inventory for all product variants',
+)]
+final class UpdateInventoryCommand extends Command
+{
+    public function __construct(private readonly InventoryUpdaterInterface $inventoryUpdater)
+    {
+        parent::__construct();
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        // todo allow the user to specify a product variant id to update
+        // todo allow the user to force the update of _ALL_ product variants regardless of the last update time
+        $this->inventoryUpdater->updateAll();
+
+        return 0;
+    }
+}
diff --git a/src/DependencyInjection/Configuration.php b/src/DependencyInjection/Configuration.php
index 1bbe6f0..241c28f 100644
--- a/src/DependencyInjection/Configuration.php
+++ b/src/DependencyInjection/Configuration.php
@@ -4,6 +4,7 @@
 
 namespace Setono\SyliusPeakPlugin\DependencyInjection;
 
+use Setono\SyliusPeakPlugin\Model\InventoryUpdate;
 use Setono\SyliusPeakPlugin\Model\RegisteredWebhooks;
 use Setono\SyliusPeakPlugin\Model\UploadOrderRequest;
 use Sylius\Component\Resource\Factory\Factory;
@@ -42,6 +43,20 @@ private function addResourcesSection(ArrayNodeDefinition $node): void
                 ->arrayNode('resources')
                     ->addDefaultsIfNotSet()
                     ->children()
+                        ->arrayNode('inventory_update')
+                            ->addDefaultsIfNotSet()
+                            ->children()
+                                ->variableNode('options')->end()
+                                ->arrayNode('classes')
+                                    ->addDefaultsIfNotSet()
+                                    ->children()
+                                        ->scalarNode('model')->defaultValue(InventoryUpdate::class)->cannotBeEmpty()->end()
+                                        ->scalarNode('repository')->cannotBeEmpty()->end()
+                                        ->scalarNode('factory')->defaultValue(Factory::class)->end()
+                                    ->end()
+                                ->end()
+                            ->end()
+                        ->end()
                         ->arrayNode('registered_webhooks')
                             ->addDefaultsIfNotSet()
                             ->children()
diff --git a/src/DependencyInjection/SetonoSyliusPeakExtension.php b/src/DependencyInjection/SetonoSyliusPeakExtension.php
index ad46aad..0993160 100644
--- a/src/DependencyInjection/SetonoSyliusPeakExtension.php
+++ b/src/DependencyInjection/SetonoSyliusPeakExtension.php
@@ -6,6 +6,7 @@
 
 use Setono\SyliusPeakPlugin\DataMapper\SalesOrderDataMapperInterface;
 use Setono\SyliusPeakPlugin\WebhookHandler\WebhookHandlerInterface;
+use Setono\SyliusPeakPlugin\Workflow\InventoryUpdateWorkflow;
 use Setono\SyliusPeakPlugin\Workflow\UploadOrderRequestWorkflow;
 use Sylius\Bundle\ResourceBundle\DependencyInjection\Extension\AbstractResourceExtension;
 use Sylius\Bundle\ResourceBundle\SyliusResourceBundle;
@@ -58,7 +59,7 @@ public function prepend(ContainerBuilder $container): void
                     ],
                 ],
             ],
-            'workflows' => UploadOrderRequestWorkflow::getConfig(),
+            'workflows' => UploadOrderRequestWorkflow::getConfig() + InventoryUpdateWorkflow::getConfig(),
         ]);
     }
 }
diff --git a/src/EventSubscriber/Workflow/InventoryUpdate/CompleteSubscriber.php b/src/EventSubscriber/Workflow/InventoryUpdate/CompleteSubscriber.php
new file mode 100644
index 0000000..5fa619d
--- /dev/null
+++ b/src/EventSubscriber/Workflow/InventoryUpdate/CompleteSubscriber.php
@@ -0,0 +1,34 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Setono\SyliusPeakPlugin\EventSubscriber\Workflow\InventoryUpdate;
+
+use Setono\SyliusPeakPlugin\Model\InventoryUpdateInterface;
+use Setono\SyliusPeakPlugin\Workflow\InventoryUpdateWorkflow;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Workflow\Event\CompletedEvent;
+use Webmozart\Assert\Assert;
+
+final class CompleteSubscriber implements EventSubscriberInterface
+{
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            sprintf('workflow.%s.completed.%s', InventoryUpdateWorkflow::NAME, InventoryUpdateWorkflow::TRANSITION_COMPLETE) => 'set',
+        ];
+    }
+
+    public function set(CompletedEvent $event): void
+    {
+        /** @var InventoryUpdateInterface|object $inventoryUpdate */
+        $inventoryUpdate = $event->getSubject();
+        Assert::isInstanceOf($inventoryUpdate, InventoryUpdateInterface::class);
+
+        $inventoryUpdate->setCompletedAt(new \DateTimeImmutable());
+
+        if (!$inventoryUpdate->hasErrors()) {
+            $inventoryUpdate->setNextUpdateThreshold($inventoryUpdate->getProcessingStartedAt());
+        }
+    }
+}
diff --git a/src/EventSubscriber/Workflow/InventoryUpdate/ProcessSubscriber.php b/src/EventSubscriber/Workflow/InventoryUpdate/ProcessSubscriber.php
new file mode 100644
index 0000000..daf2ee0
--- /dev/null
+++ b/src/EventSubscriber/Workflow/InventoryUpdate/ProcessSubscriber.php
@@ -0,0 +1,30 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Setono\SyliusPeakPlugin\EventSubscriber\Workflow\InventoryUpdate;
+
+use Setono\SyliusPeakPlugin\Model\InventoryUpdateInterface;
+use Setono\SyliusPeakPlugin\Workflow\InventoryUpdateWorkflow;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Workflow\Event\CompletedEvent;
+use Webmozart\Assert\Assert;
+
+final class ProcessSubscriber implements EventSubscriberInterface
+{
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            sprintf('workflow.%s.completed.%s', InventoryUpdateWorkflow::NAME, InventoryUpdateWorkflow::TRANSITION_PROCESS) => 'set',
+        ];
+    }
+
+    public function set(CompletedEvent $event): void
+    {
+        /** @var InventoryUpdateInterface|object $inventoryUpdate */
+        $inventoryUpdate = $event->getSubject();
+        Assert::isInstanceOf($inventoryUpdate, InventoryUpdateInterface::class);
+
+        $inventoryUpdate->setProcessingStartedAt(new \DateTimeImmutable());
+    }
+}
diff --git a/src/EventSubscriber/Workflow/InventoryUpdate/ResetSubscriber.php b/src/EventSubscriber/Workflow/InventoryUpdate/ResetSubscriber.php
new file mode 100644
index 0000000..32927ad
--- /dev/null
+++ b/src/EventSubscriber/Workflow/InventoryUpdate/ResetSubscriber.php
@@ -0,0 +1,33 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Setono\SyliusPeakPlugin\EventSubscriber\Workflow\InventoryUpdate;
+
+use Setono\SyliusPeakPlugin\Model\InventoryUpdateInterface;
+use Setono\SyliusPeakPlugin\Workflow\InventoryUpdateWorkflow;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Workflow\Event\CompletedEvent;
+use Webmozart\Assert\Assert;
+
+final class ResetSubscriber implements EventSubscriberInterface
+{
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            sprintf('workflow.%s.completed.%s', InventoryUpdateWorkflow::NAME, InventoryUpdateWorkflow::TRANSITION_RESET) => 'reset',
+        ];
+    }
+
+    public function reset(CompletedEvent $event): void
+    {
+        /** @var InventoryUpdateInterface|object $inventoryUpdate */
+        $inventoryUpdate = $event->getSubject();
+        Assert::isInstanceOf($inventoryUpdate, InventoryUpdateInterface::class);
+
+        $inventoryUpdate->setCompletedAt(null);
+        $inventoryUpdate->setProcessingStartedAt(null);
+        $inventoryUpdate->setWarnings(null);
+        $inventoryUpdate->setErrors(null);
+    }
+}
diff --git a/src/Message/Command/UpdateInventory.php b/src/Message/Command/UpdateInventory.php
index b7e09cc..d5f1614 100644
--- a/src/Message/Command/UpdateInventory.php
+++ b/src/Message/Command/UpdateInventory.php
@@ -6,19 +6,28 @@
 
 use Sylius\Component\Core\Model\ProductVariantInterface;
 
-/**
- * Will update the inventory for a product variant
- */
 final class UpdateInventory implements CommandInterface
 {
-    public int $productVariant;
+    public ?int $productVariant = null;
 
-    public function __construct(int|ProductVariantInterface $productVariant)
+    private function __construct()
+    {
+    }
+
+    public static function for(int|ProductVariantInterface $productVariant): self
     {
         if ($productVariant instanceof ProductVariantInterface) {
             $productVariant = (int) $productVariant->getId();
         }
 
-        $this->productVariant = $productVariant;
+        $command = new self();
+        $command->productVariant = $productVariant;
+
+        return $command;
+    }
+
+    public static function forAll(): self
+    {
+        return new self();
     }
 }
diff --git a/src/Message/CommandHandler/UpdateInventoryHandler.php b/src/Message/CommandHandler/UpdateInventoryHandler.php
index 49b016c..e92a4ab 100644
--- a/src/Message/CommandHandler/UpdateInventoryHandler.php
+++ b/src/Message/CommandHandler/UpdateInventoryHandler.php
@@ -4,9 +4,8 @@
 
 namespace Setono\SyliusPeakPlugin\Message\CommandHandler;
 
-use Setono\PeakWMS\Client\ClientInterface;
-use Setono\PeakWMS\DataTransferObject\Product\Product;
 use Setono\SyliusPeakPlugin\Message\Command\UpdateInventory;
+use Setono\SyliusPeakPlugin\Updater\InventoryUpdaterInterface;
 use Sylius\Component\Core\Model\ProductVariantInterface;
 use Sylius\Component\Core\Repository\ProductVariantRepositoryInterface;
 use Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException;
@@ -14,48 +13,24 @@
 final class UpdateInventoryHandler
 {
     public function __construct(
-        private readonly ClientInterface $client,
         private readonly ProductVariantRepositoryInterface $productVariantRepository,
+        private readonly InventoryUpdaterInterface $inventoryUpdater,
     ) {
     }
 
     public function __invoke(UpdateInventory $message): void
     {
-        $productVariant = $this->productVariantRepository->find($message->productVariant);
-        if (!$productVariant instanceof ProductVariantInterface) {
-            throw new UnrecoverableMessageHandlingException(sprintf('Product variant with id %d not found', $message->productVariant));
-        }
-
-        $productCode = $productVariant->getProduct()?->getCode();
-        $variantCode = $productVariant->getCode();
-
-        if (null === $productCode || null === $variantCode) {
-            throw new UnrecoverableMessageHandlingException(sprintf('Product variant with id %d does not have a product code or variant code', $message->productVariant));
-        }
-
-        $collection = $this
-            ->client
-            ->product()
-            ->getByProductId($productCode)
-            ->filter(fn (Product $product) => $product->variantId === $variantCode)
-        ;
+        if (null === $message->productVariant) {
+            $this->inventoryUpdater->updateAll();
 
-        if (count($collection) === 0) {
-            throw new UnrecoverableMessageHandlingException(sprintf('The product with id %s does not have a variant with id/code %s', $productCode, $variantCode));
+            return;
         }
 
-        if (count($collection) > 1) {
-            throw new UnrecoverableMessageHandlingException(sprintf('The product with id %s has multiple products with the same variant id/code', $productCode));
-        }
-
-        $peakProduct = $collection[0];
-
-        if (null === $peakProduct->availableToSell) {
-            throw new UnrecoverableMessageHandlingException(sprintf('The product with id %s and variant id/code %s does not have an availableToSell value', $productCode, $variantCode));
+        $productVariant = $this->productVariantRepository->find($message->productVariant);
+        if (!$productVariant instanceof ProductVariantInterface) {
+            throw new UnrecoverableMessageHandlingException(sprintf('Product variant with id %d does not exist', $message->productVariant));
         }
 
-        $productVariant->setOnHand($peakProduct->availableToSell + (int) $productVariant->getOnHold());
-
-        $this->productVariantRepository->add($productVariant);
+        $this->inventoryUpdater->update($productVariant);
     }
 }
diff --git a/src/Model/InventoryUpdate.php b/src/Model/InventoryUpdate.php
new file mode 100644
index 0000000..28288a2
--- /dev/null
+++ b/src/Model/InventoryUpdate.php
@@ -0,0 +1,141 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Setono\SyliusPeakPlugin\Model;
+
+class InventoryUpdate implements InventoryUpdateInterface
+{
+    protected ?int $id = null;
+
+    protected int $version = 1;
+
+    protected string $state = InventoryUpdateInterface::STATE_PENDING;
+
+    protected ?\DateTimeInterface $processingStartedAt = null;
+
+    protected ?\DateTimeInterface $completedAt = null;
+
+    protected ?\DateTimeInterface $nextUpdateThreshold = null;
+
+    protected int $productsProcessed = 0;
+
+    /** @var list<string>|null */
+    protected ?array $warnings = [];
+
+    /** @var list<string>|null */
+    protected ?array $errors = [];
+
+    public function getId(): ?int
+    {
+        return $this->id;
+    }
+
+    public function getVersion(): int
+    {
+        return $this->version;
+    }
+
+    public function setVersion(?int $version): void
+    {
+        $this->version = (int) $version;
+    }
+
+    public function getState(): string
+    {
+        return $this->state;
+    }
+
+    public function setState(string $state): void
+    {
+        $this->state = $state;
+    }
+
+    public function getProcessingStartedAt(): ?\DateTimeInterface
+    {
+        return $this->processingStartedAt;
+    }
+
+    public function setProcessingStartedAt(?\DateTimeInterface $processingStartedAt): void
+    {
+        $this->processingStartedAt = $processingStartedAt;
+    }
+
+    public function getCompletedAt(): ?\DateTimeInterface
+    {
+        return $this->completedAt;
+    }
+
+    public function setCompletedAt(?\DateTimeInterface $completedAt): void
+    {
+        $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;
+    }
+
+    public function setProductsProcessed(int $productsProcessed): void
+    {
+        $this->productsProcessed = $productsProcessed;
+    }
+
+    public function addWarning(string $warning): void
+    {
+        $this->warnings[] = $warning;
+    }
+
+    public function getWarnings(): array
+    {
+        return $this->warnings ?? [];
+    }
+
+    public function setWarnings(?array $warnings): void
+    {
+        if ([] === $warnings) {
+            $warnings = null;
+        }
+
+        $this->warnings = $warnings;
+    }
+
+    public function hasWarnings(): bool
+    {
+        return [] !== $this->warnings;
+    }
+
+    public function addError(string $error): void
+    {
+        $this->errors[] = $error;
+    }
+
+    public function getErrors(): array
+    {
+        return $this->errors ?? [];
+    }
+
+    public function setErrors(?array $errors): void
+    {
+        if ([] === $errors) {
+            $errors = null;
+        }
+
+        $this->errors = $errors;
+    }
+
+    public function hasErrors(): bool
+    {
+        return [] !== $this->errors;
+    }
+}
diff --git a/src/Model/InventoryUpdateInterface.php b/src/Model/InventoryUpdateInterface.php
new file mode 100644
index 0000000..7bac5ac
--- /dev/null
+++ b/src/Model/InventoryUpdateInterface.php
@@ -0,0 +1,76 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Setono\SyliusPeakPlugin\Model;
+
+use Sylius\Component\Resource\Model\ResourceInterface;
+use Sylius\Component\Resource\Model\VersionedInterface;
+
+interface InventoryUpdateInterface extends ResourceInterface, VersionedInterface
+{
+    final public const STATE_PENDING = 'pending';
+
+    final public const STATE_PROCESSING = 'processing';
+
+    final public const STATE_COMPLETED = 'completed';
+
+    public function getId(): ?int;
+
+    public function getState(): string;
+
+    public function setState(string $state): void;
+
+    /**
+     * This is the time when the inventory update started processing. If null, it means that the inventory update has never been processed.
+     */
+    public function getProcessingStartedAt(): ?\DateTimeInterface;
+
+    public function setProcessingStartedAt(?\DateTimeInterface $processingStartedAt): void;
+
+    /**
+     * This is the time when the inventory update was completed. If null, it means that the inventory update has not been completed yet.
+     */
+    public function getCompletedAt(): ?\DateTimeInterface;
+
+    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;
+
+    public function addWarning(string $warning): void;
+
+    /**
+     * @return list<string>
+     */
+    public function getWarnings(): array;
+
+    /**
+     * @param list<string>|null $warnings
+     */
+    public function setWarnings(?array $warnings): void;
+
+    public function hasWarnings(): bool;
+
+    /**
+     * @return list<string>
+     */
+    public function getErrors(): array;
+
+    public function addError(string $error): void;
+
+    /**
+     * @param list<string>|null $errors
+     */
+    public function setErrors(?array $errors): void;
+
+    public function hasErrors(): bool;
+}
diff --git a/src/Provider/InventoryUpdateProvider.php b/src/Provider/InventoryUpdateProvider.php
new file mode 100644
index 0000000..d8fb0b2
--- /dev/null
+++ b/src/Provider/InventoryUpdateProvider.php
@@ -0,0 +1,43 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Setono\SyliusPeakPlugin\Provider;
+
+use Doctrine\Persistence\ManagerRegistry;
+use Setono\Doctrine\ORMTrait;
+use Setono\SyliusPeakPlugin\Model\InventoryUpdateInterface;
+use Sylius\Component\Resource\Factory\FactoryInterface;
+use Webmozart\Assert\Assert;
+
+final class InventoryUpdateProvider implements InventoryUpdateProviderInterface
+{
+    use ORMTrait;
+
+    public function __construct(ManagerRegistry $managerRegistry, private readonly FactoryInterface $factory)
+    {
+        $this->managerRegistry = $managerRegistry;
+    }
+
+    public function getInventoryUpdate(): InventoryUpdateInterface
+    {
+        $inventoryUpdates = $this->managerRegistry->getRepository(InventoryUpdateInterface::class)->findAll();
+        $c = count($inventoryUpdates);
+
+        if ($c > 1) {
+            throw new \RuntimeException(sprintf(
+                'Expected to find zero or one inventory update, but found %d',
+                $c,
+            ));
+        }
+
+        if ($c === 1) {
+            return $inventoryUpdates[0];
+        }
+
+        $inventoryUpdate = $this->factory->createNew();
+        Assert::isInstanceOf($inventoryUpdate, InventoryUpdateInterface::class);
+
+        return $inventoryUpdate;
+    }
+}
diff --git a/src/Provider/InventoryUpdateProviderInterface.php b/src/Provider/InventoryUpdateProviderInterface.php
new file mode 100644
index 0000000..5cc26b0
--- /dev/null
+++ b/src/Provider/InventoryUpdateProviderInterface.php
@@ -0,0 +1,15 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Setono\SyliusPeakPlugin\Provider;
+
+use Setono\SyliusPeakPlugin\Model\InventoryUpdateInterface;
+
+interface InventoryUpdateProviderInterface
+{
+    /**
+     * Will return an inventory update entity. If one doesn't exist in the database yet, this provider will create it
+     */
+    public function getInventoryUpdate(): InventoryUpdateInterface;
+}
diff --git a/src/Resources/config/doctrine/model/InventoryUpdate.orm.xml b/src/Resources/config/doctrine/model/InventoryUpdate.orm.xml
new file mode 100644
index 0000000..5a6c087
--- /dev/null
+++ b/src/Resources/config/doctrine/model/InventoryUpdate.orm.xml
@@ -0,0 +1,22 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<doctrine-mapping xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+                  xmlns="http://doctrine-project.org/schemas/orm/doctrine-mapping"
+                  xsi:schemaLocation="http://doctrine-project.org/schemas/orm/doctrine-mapping
+                                      http://doctrine-project.org/schemas/orm/doctrine-mapping.xsd">
+    <mapped-superclass name="Setono\SyliusPeakPlugin\Model\InventoryUpdate"
+                       table="setono_sylius_peak_wms__inventory_update">
+        <id name="id" type="integer">
+            <generator strategy="AUTO"/>
+        </id>
+
+        <field name="version" type="integer" version="true"/>
+        <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"/>
+    </mapped-superclass>
+</doctrine-mapping>
diff --git a/src/Resources/config/services.xml b/src/Resources/config/services.xml
index 3d1be15..83626b0 100644
--- a/src/Resources/config/services.xml
+++ b/src/Resources/config/services.xml
@@ -13,6 +13,7 @@
         <import resource="services/processor.xml"/>
         <import resource="services/provider.xml"/>
         <import resource="services/registrar.xml"/>
+        <import resource="services/updater.xml"/>
         <import resource="services/webhook_handler.xml"/>
     </imports>
 </container>
diff --git a/src/Resources/config/services/client.xml b/src/Resources/config/services/client.xml
index 094f9fb..5100b63 100644
--- a/src/Resources/config/services/client.xml
+++ b/src/Resources/config/services/client.xml
@@ -2,6 +2,8 @@
 <container xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://symfony.com/schema/dic/services"
            xsi:schemaLocation="http://symfony.com/schema/dic/services http://symfony.com/schema/dic/services/services-1.0.xsd">
     <services>
+        <service id="Setono\PeakWMS\Client\ClientInterface" alias="setono_sylius_peak.client"/>
+
         <service id="setono_sylius_peak.client" class="Setono\PeakWMS\Client\Client">
             <argument>%setono_sylius_peak.api_key%</argument>
 
diff --git a/src/Resources/config/services/command.xml b/src/Resources/config/services/command.xml
index dc02511..1fe2cae 100644
--- a/src/Resources/config/services/command.xml
+++ b/src/Resources/config/services/command.xml
@@ -2,18 +2,22 @@
 <container xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://symfony.com/schema/dic/services"
            xsi:schemaLocation="http://symfony.com/schema/dic/services http://symfony.com/schema/dic/services/services-1.0.xsd">
     <services>
-        <service id="setono_sylius_peak.command.upload_orders"
-                 class="Setono\SyliusPeakPlugin\Command\ProcessCommand">
+        <service id="Setono\SyliusPeakPlugin\Command\ProcessCommand">
             <argument type="service" id="setono_sylius_peak.processor.upload_order_request"/>
 
             <tag name="console.command"/>
         </service>
 
-        <service id="setono_sylius_peak.command.register_webhooks"
-                 class="Setono\SyliusPeakPlugin\Command\RegisterWebhooksCommand">
+        <service id="Setono\SyliusPeakPlugin\Command\RegisterWebhooksCommand">
             <argument type="service" id="setono_sylius_peak.command_bus"/>
 
             <tag name="console.command"/>
         </service>
+
+        <service id="Setono\SyliusPeakPlugin\Command\UpdateInventoryCommand">
+            <argument type="service" id="Setono\SyliusPeakPlugin\Updater\InventoryUpdaterInterface"/>
+
+            <tag name="console.command"/>
+        </service>
     </services>
 </container>
diff --git a/src/Resources/config/services/event_subscriber.xml b/src/Resources/config/services/event_subscriber.xml
index eb6551a..1d57803 100644
--- a/src/Resources/config/services/event_subscriber.xml
+++ b/src/Resources/config/services/event_subscriber.xml
@@ -26,5 +26,18 @@
 
             <tag name="kernel.event_subscriber"/>
         </service>
+
+        <!-- Workflow -->
+        <service id="Setono\SyliusPeakPlugin\EventSubscriber\Workflow\InventoryUpdate\CompleteSubscriber">
+            <tag name="kernel.event_subscriber"/>
+        </service>
+
+        <service id="Setono\SyliusPeakPlugin\EventSubscriber\Workflow\InventoryUpdate\ProcessSubscriber">
+            <tag name="kernel.event_subscriber"/>
+        </service>
+
+        <service id="Setono\SyliusPeakPlugin\EventSubscriber\Workflow\InventoryUpdate\ResetSubscriber">
+            <tag name="kernel.event_subscriber"/>
+        </service>
     </services>
 </container>
diff --git a/src/Resources/config/services/message.xml b/src/Resources/config/services/message.xml
index f608de5..127f983 100644
--- a/src/Resources/config/services/message.xml
+++ b/src/Resources/config/services/message.xml
@@ -19,5 +19,12 @@
 
             <tag name="messenger.message_handler"/>
         </service>
+
+        <service id="Setono\SyliusPeakPlugin\Message\CommandHandler\UpdateInventoryHandler">
+            <argument type="service" id="sylius.repository.product_variant"/>
+            <argument type="service" id="Setono\SyliusPeakPlugin\Updater\InventoryUpdaterInterface"/>
+
+            <tag name="messenger.message_handler"/>
+        </service>
     </services>
 </container>
diff --git a/src/Resources/config/services/provider.xml b/src/Resources/config/services/provider.xml
index 602e0ee..84f38df 100644
--- a/src/Resources/config/services/provider.xml
+++ b/src/Resources/config/services/provider.xml
@@ -14,5 +14,13 @@
             <argument type="service" id="doctrine"/>
             <argument>%sylius.model.product_variant.class%</argument>
         </service>
+
+        <service id="Setono\SyliusPeakPlugin\Provider\InventoryUpdateProviderInterface"
+                 alias="Setono\SyliusPeakPlugin\Provider\InventoryUpdateProvider"/>
+
+        <service id="Setono\SyliusPeakPlugin\Provider\InventoryUpdateProvider">
+            <argument type="service" id="doctrine"/>
+            <argument type="service" id="setono_sylius_peak.factory.inventory_update"/>
+        </service>
     </services>
 </container>
diff --git a/src/Resources/config/services/updater.xml b/src/Resources/config/services/updater.xml
new file mode 100644
index 0000000..b40c760
--- /dev/null
+++ b/src/Resources/config/services/updater.xml
@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<container xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://symfony.com/schema/dic/services"
+           xsi:schemaLocation="http://symfony.com/schema/dic/services http://symfony.com/schema/dic/services/services-1.0.xsd">
+    <services>
+        <service id="Setono\SyliusPeakPlugin\Updater\InventoryUpdaterInterface"
+                 alias="Setono\SyliusPeakPlugin\Updater\InventoryUpdater"/>
+
+        <service id="Setono\SyliusPeakPlugin\Updater\InventoryUpdater">
+            <argument type="service" id="Setono\PeakWMS\Client\ClientInterface"/>
+            <argument type="service" id="Setono\SyliusPeakPlugin\Provider\InventoryUpdateProviderInterface"/>
+            <argument type="service" id="state_machine.setono_sylius_peak__inventory_update"/>
+            <argument type="service" id="doctrine"/>
+        </service>
+    </services>
+</container>
diff --git a/src/Updater/InventoryUpdater.php b/src/Updater/InventoryUpdater.php
new file mode 100644
index 0000000..ff2b7c6
--- /dev/null
+++ b/src/Updater/InventoryUpdater.php
@@ -0,0 +1,129 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Setono\SyliusPeakPlugin\Updater;
+
+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\SyliusPeakPlugin\Model\InventoryUpdateInterface;
+use Setono\SyliusPeakPlugin\Provider\InventoryUpdateProviderInterface;
+use Setono\SyliusPeakPlugin\Workflow\InventoryUpdateWorkflow;
+use Sylius\Component\Core\Model\ProductVariant;
+use Sylius\Component\Core\Model\ProductVariantInterface;
+use Symfony\Component\Workflow\WorkflowInterface;
+use Webmozart\Assert\Assert;
+
+final class InventoryUpdater implements InventoryUpdaterInterface
+{
+    use ORMTrait;
+
+    public function __construct(
+        private readonly ClientInterface $client,
+        private readonly InventoryUpdateProviderInterface $inventoryUpdateProvider,
+        private readonly WorkflowInterface $inventoryUpdateWorkflow,
+        ManagerRegistry $managerRegistry,
+    ) {
+        $this->managerRegistry = $managerRegistry;
+    }
+
+    public function update(ProductVariantInterface $productVariant): void
+    {
+        $productCode = $productVariant->getProduct()?->getCode();
+        $variantCode = $productVariant->getCode();
+
+        if (null === $productCode || null === $variantCode) {
+            throw new \RuntimeException(sprintf(
+                'Product variant with id %d does not have a product code or variant code',
+                (int) $productVariant->getId(),
+            ));
+        }
+
+        $collection = $this
+            ->client
+            ->product()
+            ->getByProductId($productCode)
+            ->filter(fn (Product $product) => $product->variantId === $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->getManager($productVariant)->flush();
+    }
+
+    // todo what happens if the transitions are not possible?
+    public function updateAll(bool $onlyUpdated = true): void
+    {
+        $inventoryUpdate = $this->inventoryUpdateProvider->getInventoryUpdate();
+        $this->getManager($inventoryUpdate)->persist($inventoryUpdate);
+
+        $this->transition($inventoryUpdate, InventoryUpdateWorkflow::TRANSITION_RESET);
+        $this->transition($inventoryUpdate, InventoryUpdateWorkflow::TRANSITION_PROCESS);
+
+        $manager = $this->getManager(ProductVariant::class);
+        $productVariantRepository = $this->getRepository(ProductVariant::class);
+
+        $i = 0;
+        $products = $this->client->product()->iterate(PageQuery::create(updatedAfter: $inventoryUpdate->getNextUpdateThreshold()));
+        foreach ($products as $product) {
+            ++$i;
+
+            if ($i % 100 === 0) {
+                $manager->flush();
+                $manager->clear();
+            }
+
+            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,
+                ));
+
+                $productVariant = $productVariantRepository->findOneBy(['code' => $product->variantId]);
+                Assert::notNull($productVariant, sprintf('Product variant with code %s does not exist', $product->variantId));
+
+                if ($product->orderedByCustomers !== $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,
+                        (int) $productVariant->getOnHold(),
+                        (int) $product->orderedByCustomers,
+                    ));
+                }
+
+                $this->map($product, $productVariant);
+            } catch (\InvalidArgumentException $e) {
+                $inventoryUpdate->addError($e->getMessage());
+            }
+        }
+
+        $inventoryUpdate->setProductsProcessed($i);
+
+        $manager->flush();
+
+        $this->transition($inventoryUpdate, InventoryUpdateWorkflow::TRANSITION_COMPLETE);
+    }
+
+    private function transition(InventoryUpdateInterface $inventoryUpdate, string $transition): void
+    {
+        $this->inventoryUpdateWorkflow->apply($inventoryUpdate, $transition);
+
+        $this->getManager($inventoryUpdate)->flush();
+    }
+
+    private function map(Product $product, ProductVariantInterface $productVariant): void
+    {
+        $productVariant->setOnHand((int) $product->availableToSell + (int) $productVariant->getOnHold());
+    }
+}
diff --git a/src/Updater/InventoryUpdaterInterface.php b/src/Updater/InventoryUpdaterInterface.php
new file mode 100644
index 0000000..26a4a0a
--- /dev/null
+++ b/src/Updater/InventoryUpdaterInterface.php
@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Setono\SyliusPeakPlugin\Updater;
+
+use Sylius\Component\Core\Model\ProductVariantInterface;
+
+interface InventoryUpdaterInterface
+{
+    /**
+     * Will update the inventory for a single product variant
+     */
+    public function update(ProductVariantInterface $productVariant): void;
+
+    /**
+     * Will update the inventory for all product variants
+     *
+     * @param bool $onlyUpdated If true, only the product variants that have been updated since the last update will be updated
+     */
+    public function updateAll(bool $onlyUpdated = true): void;
+}
diff --git a/src/WebhookHandler/StockAdjustmentWebhookHandler.php b/src/WebhookHandler/StockAdjustmentWebhookHandler.php
index 094c650..78ec845 100644
--- a/src/WebhookHandler/StockAdjustmentWebhookHandler.php
+++ b/src/WebhookHandler/StockAdjustmentWebhookHandler.php
@@ -29,7 +29,7 @@ public function handle(object $data): void
             throw new \InvalidArgumentException(sprintf('Product variant with id/code "%s" not found', (string) $data->variantId));
         }
 
-        $this->commandBus->dispatch(new UpdateInventory($productVariant));
+        $this->commandBus->dispatch(UpdateInventory::for($productVariant));
     }
 
     /**
diff --git a/src/Workflow/InventoryUpdateWorkflow.php b/src/Workflow/InventoryUpdateWorkflow.php
new file mode 100644
index 0000000..4afee97
--- /dev/null
+++ b/src/Workflow/InventoryUpdateWorkflow.php
@@ -0,0 +1,86 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Setono\SyliusPeakPlugin\Workflow;
+
+use Setono\SyliusPeakPlugin\Model\InventoryUpdateInterface;
+use Symfony\Component\Workflow\Transition;
+
+final class InventoryUpdateWorkflow
+{
+    private const PROPERTY_NAME = 'state';
+
+    final public const NAME = 'setono_sylius_peak__inventory_update';
+
+    final public const TRANSITION_PROCESS = 'process';
+
+    final public const TRANSITION_COMPLETE = 'complete';
+
+    final public const TRANSITION_RESET = 'reset';
+
+    private function __construct()
+    {
+    }
+
+    /**
+     * @return array<array-key, string>
+     */
+    public static function getStates(): array
+    {
+        return [
+            InventoryUpdateInterface::STATE_PENDING,
+            InventoryUpdateInterface::STATE_PROCESSING,
+            InventoryUpdateInterface::STATE_COMPLETED,
+        ];
+    }
+
+    public static function getConfig(): array
+    {
+        $transitions = [];
+        foreach (self::getTransitions() as $transition) {
+            $transitions[$transition->getName()] = [
+                'from' => $transition->getFroms(),
+                'to' => $transition->getTos(),
+            ];
+        }
+
+        return [
+            self::NAME => [
+                'type' => 'state_machine',
+                'marking_store' => [
+                    'type' => 'method',
+                    'property' => self::PROPERTY_NAME,
+                ],
+                'supports' => InventoryUpdateInterface::class,
+                'initial_marking' => InventoryUpdateInterface::STATE_PENDING,
+                'places' => self::getStates(),
+                'transitions' => $transitions,
+            ],
+        ];
+    }
+
+    /**
+     * @return array<array-key, Transition>
+     */
+    public static function getTransitions(): array
+    {
+        return [
+            new Transition(
+                self::TRANSITION_PROCESS,
+                InventoryUpdateInterface::STATE_PENDING,
+                InventoryUpdateInterface::STATE_PROCESSING,
+            ),
+            new Transition(
+                self::TRANSITION_COMPLETE,
+                InventoryUpdateInterface::STATE_PROCESSING,
+                InventoryUpdateInterface::STATE_COMPLETED,
+            ),
+            new Transition(
+                self::TRANSITION_RESET,
+                [InventoryUpdateInterface::STATE_PENDING, InventoryUpdateInterface::STATE_COMPLETED],
+                InventoryUpdateInterface::STATE_PENDING,
+            ),
+        ];
+    }
+}