vendor/shopware/core/Content/Product/DataAbstractionLayer/StockUpdater.php line 62

  1. <?php declare(strict_types=1);
  3. namespace Shopware\Core\Content\Product\DataAbstractionLayer;
  5. use Doctrine\DBAL\ArrayParameterType;
  6. use Doctrine\DBAL\Connection;
  7. use Shopware\Core\Checkout\Cart\Event\CheckoutOrderPlacedEvent;
  8. use Shopware\Core\Checkout\Cart\LineItem\LineItem;
  9. use Shopware\Core\Checkout\Order\Aggregate\OrderLineItem\OrderLineItemDefinition;
  10. use Shopware\Core\Checkout\Order\OrderEvents;
  11. use Shopware\Core\Checkout\Order\OrderStates;
  12. use Shopware\Core\Content\Product\DataAbstractionLayer\StockUpdate\StockUpdateFilterProvider;
  13. use Shopware\Core\Content\Product\Events\ProductNoLongerAvailableEvent;
  14. use Shopware\Core\Defaults;
  15. use Shopware\Core\Framework\Context;
  16. use Shopware\Core\Framework\DataAbstractionLayer\Doctrine\RetryableQuery;
  17. use Shopware\Core\Framework\DataAbstractionLayer\EntityWriteResult;
  18. use Shopware\Core\Framework\DataAbstractionLayer\Event\EntityWrittenEvent;
  19. use Shopware\Core\Framework\DataAbstractionLayer\Write\Command\ChangeSetAware;
  20. use Shopware\Core\Framework\DataAbstractionLayer\Write\Command\DeleteCommand;
  21. use Shopware\Core\Framework\DataAbstractionLayer\Write\Command\WriteCommand;
  22. use Shopware\Core\Framework\DataAbstractionLayer\Write\Validation\PreWriteValidationEvent;
  23. use Shopware\Core\Framework\Log\Package;
  24. use Shopware\Core\Framework\Uuid\Uuid;
  25. use Shopware\Core\Profiling\Profiler;
  26. use Shopware\Core\System\StateMachine\Event\StateMachineTransitionEvent;
  27. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  28. use Symfony\Contracts\EventDispatcher\EventDispatcherInterface;
  30. /**
  31.  * @internal
  32.  */
  33. #[Package('core')]
  34. class StockUpdater implements EventSubscriberInterface
  35. {
  36.     /**
  37.      * @internal
  38.      */
  39.     public function __construct(
  40.         private readonly Connection $connection,
  41.         private readonly EventDispatcherInterface $dispatcher,
  42.         private readonly StockUpdateFilterProvider $stockUpdateFilter
  43.     ) {
  44.     }
  46.     /**
  47.      * Returns a list of custom business events to listen where the product maybe changed
  48.      *
  49.      * @return array<string, string|array{0: string, 1: int}|list<array{0: string, 1?: int}>>
  50.      */
  51.     public static function getSubscribedEvents(): array
  52.     {
  53.         return [
  54.             CheckoutOrderPlacedEvent::class => 'orderPlaced',
  55.             StateMachineTransitionEvent::class => 'stateChanged',
  56.             PreWriteValidationEvent::class => 'triggerChangeSet',
  57.             OrderEvents::ORDER_LINE_ITEM_WRITTEN_EVENT => 'lineItemWritten',
  58.             OrderEvents::ORDER_LINE_ITEM_DELETED_EVENT => 'lineItemWritten',
  59.         ];
  60.     }
  62.     public function triggerChangeSet(PreWriteValidationEvent $event): void
  63.     {
  64.         if ($event->getContext()->getVersionId() !== Defaults::LIVE_VERSION) {
  65.             return;
  66.         }
  68.         foreach ($event->getCommands() as $command) {
  69.             if (!$command instanceof ChangeSetAware) {
  70.                 continue;
  71.             }
  72.             /** @var ChangeSetAware&WriteCommand $command */
  73.             if ($command->getDefinition()->getEntityName() !== OrderLineItemDefinition::ENTITY_NAME) {
  74.                 continue;
  75.             }
  76.             if ($command instanceof DeleteCommand) {
  77.                 $command->requestChangeSet();
  79.                 continue;
  80.             }
  81.             /** @var WriteCommand&ChangeSetAware $command */
  82.             if ($command->hasField('referenced_id') || $command->hasField('product_id') || $command->hasField('quantity')) {
  83.                 $command->requestChangeSet();
  84.             }
  85.         }
  86.     }
  88.     public function orderPlaced(CheckoutOrderPlacedEvent $event): void
  89.     {
  90.         $ids = [];
  91.         foreach ($event->getOrder()->getLineItems() ?? [] as $lineItem) {
  92.             if ($lineItem->getType() !== LineItem::PRODUCT_LINE_ITEM_TYPE) {
  93.                 continue;
  94.             }
  96.             $referencedId = $lineItem->getReferencedId();
  97.             if (!$referencedId) {
  98.                 continue;
  99.             }
  101.             if (!\array_key_exists($referencedId, $ids)) {
  102.                 $ids[$referencedId] = 0;
  103.             }
  105.             $ids[$referencedId] += $lineItem->getQuantity();
  106.         }
  108.         $filteredIds = $this->stockUpdateFilter->filterProductIdsForStockUpdates(\array_keys($ids), $event->getContext());
  110.         $ids = \array_filter($ids, static fn (string $id) => \in_array($id, $filteredIds, true), \ARRAY_FILTER_USE_KEY);
  112.         // order placed event is a high load event. Because of the high load, we simply reduce the quantity here instead of executing the high costs `update` function
  113.         $query = new RetryableQuery(
  114.             $this->connection,
  115.             $this->connection->prepare('UPDATE product SET available_stock = available_stock - :quantity WHERE id = :id')
  116.         );
  118.         Profiler::trace('order::update-stock', static function () use ($query, $ids): void {
  119.             foreach ($ids as $id => $quantity) {
  120.                 $query->execute(['id' => Uuid::fromHexToBytes((string) $id), 'quantity' => $quantity]);
  121.             }
  122.         });
  124.         Profiler::trace('order::update-flag', function () use ($ids, $event): void {
  125.             $this->updateAvailableFlag(\array_keys($ids), $event->getContext());
  126.         });
  127.     }
  129.     /**
  130.      * If the product of an order item changed, the stocks of the old product and the new product must be updated.
  131.      */
  132.     public function lineItemWritten(EntityWrittenEvent $event): void
  133.     {
  134.         $ids = [];
  136.         // we don't want to trigger to `update` method when we are inside the order process
  137.         if ($event->getContext()->hasState('checkout-order-route')) {
  138.             return;
  139.         }
  141.         foreach ($event->getWriteResults() as $result) {
  142.             if ($result->hasPayload('referencedId') && $result->getProperty('type') === LineItem::PRODUCT_LINE_ITEM_TYPE) {
  143.                 $ids[] = $result->getProperty('referencedId');
  144.             }
  146.             if ($result->getOperation() === EntityWriteResult::OPERATION_INSERT) {
  147.                 continue;
  148.             }
  150.             $changeSet = $result->getChangeSet();
  151.             if (!$changeSet) {
  152.                 continue;
  153.             }
  155.             $type = $changeSet->getBefore('type');
  157.             if ($type !== LineItem::PRODUCT_LINE_ITEM_TYPE) {
  158.                 continue;
  159.             }
  161.             if (!$changeSet->hasChanged('referenced_id') && !$changeSet->hasChanged('quantity')) {
  162.                 continue;
  163.             }
  165.             $ids[] = $changeSet->getBefore('referenced_id');
  166.             $ids[] = $changeSet->getAfter('referenced_id');
  167.         }
  169.         $ids = array_filter(array_unique($ids));
  171.         if (empty($ids)) {
  172.             return;
  173.         }
  175.         $this->update($ids, $event->getContext());
  176.     }
  178.     public function stateChanged(StateMachineTransitionEvent $event): void
  179.     {
  180.         if ($event->getContext()->getVersionId() !== Defaults::LIVE_VERSION) {
  181.             return;
  182.         }
  184.         if ($event->getEntityName() !== 'order') {
  185.             return;
  186.         }
  188.         if ($event->getToPlace()->getTechnicalName() === OrderStates::STATE_COMPLETED) {
  189.             $products = $this->getProductsOfOrder($event->getEntityId(), $event->getContext());
  191.             $this->updateStockAndSales($products, -1);
  193.             return;
  194.         }
  196.         if ($event->getFromPlace()->getTechnicalName() === OrderStates::STATE_COMPLETED) {
  197.             $products = $this->getProductsOfOrder($event->getEntityId(), $event->getContext());
  199.             $this->updateStockAndSales($products, +1);
  201.             return;
  202.         }
  204.         if ($event->getToPlace()->getTechnicalName() === OrderStates::STATE_CANCELLED || $event->getFromPlace()->getTechnicalName() === OrderStates::STATE_CANCELLED) {
  205.             $products = $this->getProductsOfOrder($event->getEntityId(), $event->getContext());
  207.             $ids = array_column($products, 'referenced_id');
  209.             $this->updateAvailableStockAndSales($ids, $event->getContext());
  211.             $this->updateAvailableFlag($ids, $event->getContext());
  212.         }
  213.     }
  215.     /**
  216.      * @param list<string> $ids
  217.      */
  218.     public function update(array $ids, Context $context): void
  219.     {
  220.         if ($context->getVersionId() !== Defaults::LIVE_VERSION) {
  221.             return;
  222.         }
  224.         $ids = $this->stockUpdateFilter->filterProductIdsForStockUpdates($ids, $context);
  226.         $this->updateAvailableStockAndSales($ids, $context);
  228.         $this->updateAvailableFlag($ids, $context);
  229.     }
  231.     /**
  232.      * @param list<string> $ids
  233.      */
  234.     private function updateAvailableStockAndSales(array $ids, Context $context): void
  235.     {
  236.         $ids = array_filter(array_keys(array_flip($ids)));
  238.         if (empty($ids)) {
  239.             return;
  240.         }
  242.         $sql = '
  243. SELECT LOWER(HEX(order_line_item.product_id)) as product_id,
  244.     IFNULL(
  245.         SUM(IF(state_machine_state.technical_name = :completed_state, 0, order_line_item.quantity)),
  246.         0
  247.     ) as open_quantity,
  249.     IFNULL(
  250.         SUM(IF(state_machine_state.technical_name = :completed_state, order_line_item.quantity, 0)),
  251.         0
  252.     ) as sales_quantity
  254. FROM order_line_item
  255.     INNER JOIN `order`
  256.         ON `order`.id = order_line_item.order_id
  257.         AND `order`.version_id = order_line_item.order_version_id
  258.     INNER JOIN state_machine_state
  259.         ON state_machine_state.id = `order`.state_id
  260.         AND state_machine_state.technical_name <> :cancelled_state
  262. WHERE order_line_item.product_id IN (:ids)
  263.     AND order_line_item.type = :type
  264.     AND order_line_item.version_id = :version
  265.     AND order_line_item.product_id IS NOT NULL
  266. GROUP BY product_id;
  267.         ';
  269.         $rows = $this->connection->fetchAllAssociative(
  270.             $sql,
  271.             [
  272.                 'type' => LineItem::PRODUCT_LINE_ITEM_TYPE,
  273.                 'version' => Uuid::fromHexToBytes($context->getVersionId()),
  274.                 'completed_state' => OrderStates::STATE_COMPLETED,
  275.                 'cancelled_state' => OrderStates::STATE_CANCELLED,
  276.                 'ids' => Uuid::fromHexToBytesList($ids),
  277.             ],
  278.             [
  279.                 'ids' => ArrayParameterType::STRING,
  280.             ]
  281.         );
  283.         $fallback = array_column($rows, 'product_id');
  285.         $fallback = array_diff($ids, $fallback);
  287.         $update = new RetryableQuery(
  288.             $this->connection,
  289.             $this->connection->prepare('UPDATE product SET available_stock = stock - :open_quantity, sales = :sales_quantity, updated_at = :now WHERE id = :id')
  290.         );
  292.         foreach ($fallback as $id) {
  293.             $update->execute([
  294.                 'id' => Uuid::fromHexToBytes((string) $id),
  295.                 'open_quantity' => 0,
  296.                 'sales_quantity' => 0,
  297.                 'now' => (new \DateTime())->format(Defaults::STORAGE_DATE_TIME_FORMAT),
  298.             ]);
  299.         }
  301.         foreach ($rows as $row) {
  302.             $update->execute([
  303.                 'id' => Uuid::fromHexToBytes($row['product_id']),
  304.                 'open_quantity' => $row['open_quantity'],
  305.                 'sales_quantity' => $row['sales_quantity'],
  306.                 'now' => (new \DateTime())->format(Defaults::STORAGE_DATE_TIME_FORMAT),
  307.             ]);
  308.         }
  309.     }
  311.     /**
  312.      * @param list<string> $ids
  313.      */
  314.     private function updateAvailableFlag(array $ids, Context $context): void
  315.     {
  316.         $ids = array_filter(array_unique($ids));
  318.         if (empty($ids)) {
  319.             return;
  320.         }
  322.         $bytes = Uuid::fromHexToBytesList($ids);
  324.         $sql = '
  325.             UPDATE product
  326.             LEFT JOIN product parent
  327.                 ON parent.id = product.parent_id
  328.                 AND parent.version_id = product.version_id
  330.             SET product.available = IFNULL((
  331.                 IFNULL(product.is_closeout, parent.is_closeout) * product.available_stock
  332.                 >=
  333.                 IFNULL(product.is_closeout, parent.is_closeout) * IFNULL(product.min_purchase, parent.min_purchase)
  334.             ), 0)
  335.             WHERE product.id IN (:ids)
  336.             AND product.version_id = :version
  337.         ';
  339.         RetryableQuery::retryable($this->connection, function () use ($sql, $context, $bytes): void {
  340.             $this->connection->executeStatement(
  341.                 $sql,
  342.                 ['ids' => $bytes, 'version' => Uuid::fromHexToBytes($context->getVersionId())],
  343.                 ['ids' => ArrayParameterType::STRING]
  344.             );
  345.         });
  347.         $updated = $this->connection->fetchFirstColumn(
  348.             'SELECT LOWER(HEX(id)) FROM product WHERE available = 0 AND id IN (:ids) AND product.version_id = :version',
  349.             ['ids' => $bytes, 'version' => Uuid::fromHexToBytes($context->getVersionId())],
  350.             ['ids' => ArrayParameterType::STRING]
  351.         );
  353.         if (!empty($updated)) {
  354.             $this->dispatcher->dispatch(new ProductNoLongerAvailableEvent($updated, $context));
  355.         }
  356.     }
  358.     /**
  359.      * @param list<array{referenced_id: string, quantity: string}> $products
  360.      */
  361.     private function updateStockAndSales(array $products, int $stockMultiplier): void
  362.     {
  363.         $query = new RetryableQuery(
  364.             $this->connection,
  365.             $this->connection->prepare('UPDATE product SET stock = stock + :quantity, sales = sales - :quantity WHERE id = :id AND version_id = :version')
  366.         );
  368.         foreach ($products as $product) {
  369.             $query->execute([
  370.                 'quantity' => (int) $product['quantity'] * $stockMultiplier,
  371.                 'id' => Uuid::fromHexToBytes($product['referenced_id']),
  372.                 'version' => Uuid::fromHexToBytes(Defaults::LIVE_VERSION),
  373.             ]);
  374.         }
  375.     }
  377.     /**
  378.      * @return list<array{referenced_id: string, quantity: string}>
  379.      */
  380.     private function getProductsOfOrder(string $orderId, Context $context): array
  381.     {
  382.         $query = $this->connection->createQueryBuilder();
  383.         $query->select(['referenced_id', 'quantity']);
  384.         $query->from('order_line_item');
  385.         $query->andWhere('type = :type');
  386.         $query->andWhere('order_id = :id');
  387.         $query->andWhere('version_id = :version');
  388.         $query->setParameter('id', Uuid::fromHexToBytes($orderId));
  389.         $query->setParameter('version', Uuid::fromHexToBytes(Defaults::LIVE_VERSION));
  390.         $query->setParameter('type', LineItem::PRODUCT_LINE_ITEM_TYPE);
  392.         /** @var list<array{referenced_id: string, quantity: string}> $result */
  393.         $result = $query->executeQuery()->fetchAllAssociative();
  395.         $filteredIds = $this->stockUpdateFilter->filterProductIdsForStockUpdates(\array_column($result, 'referenced_id'), $context);
  397.         return \array_filter($result, static fn (array $item) => \in_array($item['referenced_id'], $filteredIds, true));
  398.     }
  399. }