vendor/shopware/platform/src/Core/Content/Product/DataAbstractionLayer/StockUpdater.php line 59

Open in your IDE?
  1. <?php declare(strict_types=1);
  2. namespace Shopware\Core\Content\Product\DataAbstractionLayer;
  3. use Doctrine\DBAL\Connection;
  4. use Shopware\Core\Checkout\Cart\Event\CheckoutOrderPlacedEvent;
  5. use Shopware\Core\Checkout\Cart\LineItem\LineItem;
  6. use Shopware\Core\Checkout\Order\Aggregate\OrderLineItem\OrderLineItemDefinition;
  7. use Shopware\Core\Checkout\Order\OrderEvents;
  8. use Shopware\Core\Checkout\Order\OrderStates;
  9. use Shopware\Core\Content\Product\Events\ProductNoLongerAvailableEvent;
  10. use Shopware\Core\Defaults;
  11. use Shopware\Core\Framework\Context;
  12. use Shopware\Core\Framework\DataAbstractionLayer\Doctrine\RetryableQuery;
  13. use Shopware\Core\Framework\DataAbstractionLayer\EntityWriteResult;
  14. use Shopware\Core\Framework\DataAbstractionLayer\Event\EntityWrittenEvent;
  15. use Shopware\Core\Framework\DataAbstractionLayer\Write\Command\ChangeSetAware;
  16. use Shopware\Core\Framework\DataAbstractionLayer\Write\Command\DeleteCommand;
  17. use Shopware\Core\Framework\DataAbstractionLayer\Write\Command\InsertCommand;
  18. use Shopware\Core\Framework\DataAbstractionLayer\Write\Command\UpdateCommand;
  19. use Shopware\Core\Framework\DataAbstractionLayer\Write\Command\WriteCommand;
  20. use Shopware\Core\Framework\DataAbstractionLayer\Write\Validation\PreWriteValidationEvent;
  21. use Shopware\Core\Framework\Uuid\Uuid;
  22. use Shopware\Core\Profiling\Profiler;
  23. use Shopware\Core\System\StateMachine\Event\StateMachineTransitionEvent;
  24. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  25. use Symfony\Contracts\EventDispatcher\EventDispatcherInterface;
  26. class StockUpdater implements EventSubscriberInterface
  27. {
  28.     private Connection $connection;
  29.     private EventDispatcherInterface $dispatcher;
  30.     public function __construct(
  31.         Connection $connection,
  32.         EventDispatcherInterface $dispatcher
  33.     ) {
  34.         $this->connection $connection;
  35.         $this->dispatcher $dispatcher;
  36.     }
  37.     /**
  38.      * Returns a list of custom business events to listen where the product maybe changed
  39.      *
  40.      * @return array<string, string|array{0: string, 1: int}|list<array{0: string, 1?: int}>>
  41.      */
  42.     public static function getSubscribedEvents()
  43.     {
  44.         return [
  45.             CheckoutOrderPlacedEvent::class => 'orderPlaced',
  46.             StateMachineTransitionEvent::class => 'stateChanged',
  47.             PreWriteValidationEvent::class => 'triggerChangeSet',
  48.             OrderEvents::ORDER_LINE_ITEM_WRITTEN_EVENT => 'lineItemWritten',
  49.             OrderEvents::ORDER_LINE_ITEM_DELETED_EVENT => 'lineItemWritten',
  50.         ];
  51.     }
  52.     public function triggerChangeSet(PreWriteValidationEvent $event): void
  53.     {
  54.         if ($event->getContext()->getVersionId() !== Defaults::LIVE_VERSION) {
  55.             return;
  56.         }
  57.         foreach ($event->getCommands() as $command) {
  58.             if (!$command instanceof ChangeSetAware) {
  59.                 continue;
  60.             }
  61.             /** @var ChangeSetAware|InsertCommand|UpdateCommand $command */
  62.             if ($command->getDefinition()->getEntityName() !== OrderLineItemDefinition::ENTITY_NAME) {
  63.                 continue;
  64.             }
  65.             if ($command instanceof InsertCommand) {
  66.                 continue;
  67.             }
  68.             if ($command instanceof DeleteCommand) {
  69.                 $command->requestChangeSet();
  70.                 continue;
  71.             }
  72.             /** @var WriteCommand&ChangeSetAware $command */
  73.             if ($command->hasField('referenced_id') || $command->hasField('product_id') || $command->hasField('quantity')) {
  74.                 $command->requestChangeSet();
  75.             }
  76.         }
  77.     }
  78.     public function orderPlaced(CheckoutOrderPlacedEvent $event): void
  79.     {
  80.         $ids = [];
  81.         foreach ($event->getOrder()->getLineItems() as $lineItem) {
  82.             if ($lineItem->getType() !== LineItem::PRODUCT_LINE_ITEM_TYPE) {
  83.                 continue;
  84.             }
  85.             if (!\array_key_exists($lineItem->getReferencedId(), $ids)) {
  86.                 $ids[$lineItem->getReferencedId()] = 0;
  87.             }
  88.             $ids[$lineItem->getReferencedId()] += $lineItem->getQuantity();
  89.         }
  90.         // 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
  91.         $query = new RetryableQuery(
  92.             $this->connection,
  93.             $this->connection->prepare('UPDATE product SET available_stock = available_stock - :quantity WHERE id = :id')
  94.         );
  95.         Profiler::trace('order::update-stock', static function () use ($query$ids): void {
  96.             foreach ($ids as $id => $quantity) {
  97.                 $query->execute(['id' => Uuid::fromHexToBytes((string) $id), 'quantity' => $quantity]);
  98.             }
  99.         });
  100.         Profiler::trace('order::update-flag', function () use ($ids$event): void {
  101.             $this->updateAvailableFlag(\array_keys($ids), $event->getContext());
  102.         });
  103.     }
  104.     /**
  105.      * If the product of an order item changed, the stocks of the old product and the new product must be updated.
  106.      */
  107.     public function lineItemWritten(EntityWrittenEvent $event): void
  108.     {
  109.         $ids = [];
  110.         // we don't want to trigger to `update` method when we are inside the order process
  111.         if ($event->getContext()->hasState('checkout-order-route')) {
  112.             return;
  113.         }
  114.         foreach ($event->getWriteResults() as $result) {
  115.             if ($result->hasPayload('referencedId') && $result->getProperty('type') === LineItem::PRODUCT_LINE_ITEM_TYPE) {
  116.                 $ids[] = $result->getProperty('referencedId');
  117.             }
  118.             if ($result->getOperation() === EntityWriteResult::OPERATION_INSERT) {
  119.                 continue;
  120.             }
  121.             $changeSet $result->getChangeSet();
  122.             if (!$changeSet) {
  123.                 continue;
  124.             }
  125.             $type $changeSet->getBefore('type');
  126.             if ($type !== LineItem::PRODUCT_LINE_ITEM_TYPE) {
  127.                 continue;
  128.             }
  129.             if (!$changeSet->hasChanged('referenced_id') && !$changeSet->hasChanged('quantity')) {
  130.                 continue;
  131.             }
  132.             $ids[] = $changeSet->getBefore('referenced_id');
  133.             $ids[] = $changeSet->getAfter('referenced_id');
  134.         }
  135.         $ids array_filter(array_unique($ids));
  136.         if (empty($ids)) {
  137.             return;
  138.         }
  139.         $this->update($ids$event->getContext());
  140.     }
  141.     public function stateChanged(StateMachineTransitionEvent $event): void
  142.     {
  143.         if ($event->getContext()->getVersionId() !== Defaults::LIVE_VERSION) {
  144.             return;
  145.         }
  146.         if ($event->getEntityName() !== 'order') {
  147.             return;
  148.         }
  149.         if ($event->getToPlace()->getTechnicalName() === OrderStates::STATE_COMPLETED) {
  150.             $this->decreaseStock($event);
  151.             return;
  152.         }
  153.         if ($event->getFromPlace()->getTechnicalName() === OrderStates::STATE_COMPLETED) {
  154.             $this->increaseStock($event);
  155.             return;
  156.         }
  157.         if ($event->getToPlace()->getTechnicalName() === OrderStates::STATE_CANCELLED || $event->getFromPlace()->getTechnicalName() === OrderStates::STATE_CANCELLED) {
  158.             $products $this->getProductsOfOrder($event->getEntityId());
  159.             $ids array_column($products'referenced_id');
  160.             $this->updateAvailableStockAndSales($ids$event->getContext());
  161.             $this->updateAvailableFlag($ids$event->getContext());
  162.             return;
  163.         }
  164.     }
  165.     public function update(array $idsContext $context): void
  166.     {
  167.         if ($context->getVersionId() !== Defaults::LIVE_VERSION) {
  168.             return;
  169.         }
  170.         $this->updateAvailableStockAndSales($ids$context);
  171.         $this->updateAvailableFlag($ids$context);
  172.     }
  173.     private function increaseStock(StateMachineTransitionEvent $event): void
  174.     {
  175.         $products $this->getProductsOfOrder($event->getEntityId());
  176.         $ids array_column($products'referenced_id');
  177.         $this->updateStock($products, +1);
  178.         $this->updateAvailableStockAndSales($ids$event->getContext());
  179.         $this->updateAvailableFlag($ids$event->getContext());
  180.     }
  181.     private function decreaseStock(StateMachineTransitionEvent $event): void
  182.     {
  183.         $products $this->getProductsOfOrder($event->getEntityId());
  184.         $ids array_column($products'referenced_id');
  185.         $this->updateStock($products, -1);
  186.         $this->updateAvailableStockAndSales($ids$event->getContext());
  187.         $this->updateAvailableFlag($ids$event->getContext());
  188.     }
  189.     private function updateAvailableStockAndSales(array $idsContext $context): void
  190.     {
  191.         $ids array_filter(array_keys(array_flip($ids)));
  192.         if (empty($ids)) {
  193.             return;
  194.         }
  195.         $sql '
  196. SELECT LOWER(HEX(order_line_item.product_id)) as product_id,
  197.     IFNULL(
  198.         SUM(IF(state_machine_state.technical_name = :completed_state, 0, order_line_item.quantity)),
  199.         0
  200.     ) as open_quantity,
  201.     IFNULL(
  202.         SUM(IF(state_machine_state.technical_name = :completed_state, order_line_item.quantity, 0)),
  203.         0
  204.     ) as sales_quantity
  205. FROM order_line_item
  206.     INNER JOIN `order`
  207.         ON `order`.id = order_line_item.order_id
  208.         AND `order`.version_id = order_line_item.order_version_id
  209.     INNER JOIN state_machine_state
  210.         ON state_machine_state.id = `order`.state_id
  211.         AND state_machine_state.technical_name <> :cancelled_state
  212. WHERE order_line_item.product_id IN (:ids)
  213.     AND order_line_item.type = :type
  214.     AND order_line_item.version_id = :version
  215.     AND order_line_item.product_id IS NOT NULL
  216. GROUP BY product_id;
  217.         ';
  218.         $rows $this->connection->fetchAllAssociative(
  219.             $sql,
  220.             [
  221.                 'type' => LineItem::PRODUCT_LINE_ITEM_TYPE,
  222.                 'version' => Uuid::fromHexToBytes($context->getVersionId()),
  223.                 'completed_state' => OrderStates::STATE_COMPLETED,
  224.                 'cancelled_state' => OrderStates::STATE_CANCELLED,
  225.                 'ids' => Uuid::fromHexToBytesList($ids),
  226.             ],
  227.             [
  228.                 'ids' => Connection::PARAM_STR_ARRAY,
  229.             ]
  230.         );
  231.         $fallback array_column($rows'product_id');
  232.         $fallback array_diff($ids$fallback);
  233.         $update = new RetryableQuery(
  234.             $this->connection,
  235.             $this->connection->prepare('UPDATE product SET available_stock = stock - :open_quantity, sales = :sales_quantity, updated_at = :now WHERE id = :id')
  236.         );
  237.         foreach ($fallback as $id) {
  238.             $update->execute([
  239.                 'id' => Uuid::fromHexToBytes((string) $id),
  240.                 'open_quantity' => 0,
  241.                 'sales_quantity' => 0,
  242.                 'now' => (new \DateTime())->format(Defaults::STORAGE_DATE_TIME_FORMAT),
  243.             ]);
  244.         }
  245.         foreach ($rows as $row) {
  246.             $update->execute([
  247.                 'id' => Uuid::fromHexToBytes($row['product_id']),
  248.                 'open_quantity' => $row['open_quantity'],
  249.                 'sales_quantity' => $row['sales_quantity'],
  250.                 'now' => (new \DateTime())->format(Defaults::STORAGE_DATE_TIME_FORMAT),
  251.             ]);
  252.         }
  253.     }
  254.     private function updateAvailableFlag(array $idsContext $context): void
  255.     {
  256.         $ids array_filter(array_unique($ids));
  257.         if (empty($ids)) {
  258.             return;
  259.         }
  260.         $bytes Uuid::fromHexToBytesList($ids);
  261.         $sql '
  262.             UPDATE product
  263.             LEFT JOIN product parent
  264.                 ON parent.id = product.parent_id
  265.                 AND parent.version_id = product.version_id
  266.             SET product.available = IFNULL((
  267.                 IFNULL(product.is_closeout, parent.is_closeout) * product.available_stock
  268.                 >=
  269.                 IFNULL(product.is_closeout, parent.is_closeout) * IFNULL(product.min_purchase, parent.min_purchase)
  270.             ), 0)
  271.             WHERE product.id IN (:ids)
  272.             AND product.version_id = :version
  273.         ';
  274.         RetryableQuery::retryable($this->connection, function () use ($sql$context$bytes): void {
  275.             $this->connection->executeUpdate(
  276.                 $sql,
  277.                 ['ids' => $bytes'version' => Uuid::fromHexToBytes($context->getVersionId())],
  278.                 ['ids' => Connection::PARAM_STR_ARRAY]
  279.             );
  280.         });
  281.         $updated $this->connection->fetchFirstColumn(
  282.             'SELECT LOWER(HEX(id)) FROM product WHERE available = 0 AND id IN (:ids) AND product.version_id = :version',
  283.             ['ids' => $bytes'version' => Uuid::fromHexToBytes($context->getVersionId())],
  284.             ['ids' => Connection::PARAM_STR_ARRAY]
  285.         );
  286.         if (!empty($updated)) {
  287.             $this->dispatcher->dispatch(new ProductNoLongerAvailableEvent($updated$context));
  288.         }
  289.     }
  290.     private function updateStock(array $productsint $multiplier): void
  291.     {
  292.         $query = new RetryableQuery(
  293.             $this->connection,
  294.             $this->connection->prepare('UPDATE product SET stock = stock + :quantity WHERE id = :id AND version_id = :version')
  295.         );
  296.         foreach ($products as $product) {
  297.             $query->execute([
  298.                 'quantity' => (int) $product['quantity'] * $multiplier,
  299.                 'id' => Uuid::fromHexToBytes($product['referenced_id']),
  300.                 'version' => Uuid::fromHexToBytes(Defaults::LIVE_VERSION),
  301.             ]);
  302.         }
  303.     }
  304.     private function getProductsOfOrder(string $orderId): array
  305.     {
  306.         $query $this->connection->createQueryBuilder();
  307.         $query->select(['referenced_id''quantity']);
  308.         $query->from('order_line_item');
  309.         $query->andWhere('type = :type');
  310.         $query->andWhere('order_id = :id');
  311.         $query->andWhere('version_id = :version');
  312.         $query->setParameter('id'Uuid::fromHexToBytes($orderId));
  313.         $query->setParameter('version'Uuid::fromHexToBytes(Defaults::LIVE_VERSION));
  314.         $query->setParameter('type'LineItem::PRODUCT_LINE_ITEM_TYPE);
  315.         return $query->execute()->fetchAll(\PDO::FETCH_ASSOC);
  316.     }
  317. }