feat: stock

app/Amqp/Consumer/OrderGoodStockConsumer.php

@@ -4,23 +4,173 @@ declare(strict_types=1);
 
 namespace App\Amqp\Consumer;
 
+use App\Constants\Common\OrderCode;
+use App\Lib\Log;
+use App\Model\Order;
+use App\Model\OrderGood;
+use App\Model\Sku;
+use App\Service\ServiceTrait\Api\OrderTrait;
+use App\Service\ServiceTrait\Common\StockTrait;
+use Exception;
 use Hyperf\Amqp\Message\Type;
 use Hyperf\Amqp\Result;
 use Hyperf\Amqp\Annotation\Consumer;
 use Hyperf\Amqp\Message\ConsumerMessage;
+use Hyperf\Di\Annotation\Inject;
 use PhpAmqpLib\Message\AMQPMessage;
+use Psr\Container\ContainerExceptionInterface;
+use Psr\Container\NotFoundExceptionInterface;
 
 #[Consumer(exchange: 'OrderGoodStock', routingKey: 'OrderGoodStock', queue: 'OrderGoodStock.change', name: "OrderGoodStockConsumer", nums: 1)]
 class OrderGoodStockConsumer extends ConsumerMessage
 {
+    use StockTrait;
 
     /**
      * @var Type|string 消息类型
      */
     protected Type|string $type = Type::DIRECT;
 
+    /**
+     * @var Log $log
+     */
+    #[Inject]
+    protected Log $log;
+
+    /**
+     * @var OrderGood
+     */
+    #[Inject]
+    protected OrderGood $orderGoodModel;
+
+    /**
+     * @var Sku
+     */
+    #[Inject]
+    protected Sku $skuModel;
+
+    /**
+     * @var array
+     */
+    private array $orderGoodArr;
+
+    /**
+     * @var array
+     */
+    private array $skuArr;
+
+    /**
+     * @var array
+     */
+    private array $updateArr;
+
+    /**
+     * @param $data
+     * @param AMQPMessage $message
+     * @return Result
+     * @throws ContainerExceptionInterface
+     * @throws NotFoundExceptionInterface
+     */
     public function consumeMessage($data, AMQPMessage $message): Result
     {
-        return Result::ACK;
+        if (!$data['order_id'] || !$data['type']) {
+            $this->log->error('OrderGoodStockConsumer:error:NoData:'.json_encode($data));
+            return Result::ACK;
+        }
+
+        $orderId = (int)$data['order_id'];
+        $this->orderGoodArr = [];
+        $this->skuArr = [];
+
+        $this->orderGoodArr = $this->orderGoodModel->getGoodIdsByOrderId($orderId);
+        if (empty($this->orderGoodArr)) {
+            $this->log->debug('OrderGoodStockConsumer:error:NoOrderGoodData:'.json_encode($orderId));
+            return Result::ACK;
+        }
+
+        $this->skuArr = $this->skuModel->getDataArrByIds(array_column($this->orderGoodArr, 'sku_id'));
+        if (empty($this->skuArr)) {
+            $this->log->debug('OrderGoodStockConsumer:error:NoSkuData:'.json_encode(array_column($this->orderGoodArr, 'sku_id')));
+            return Result::ACK;
+        }
+
+        $this->updateArr = [];
+
+        try {
+            match ($data['type']) {
+                OrderCode::WAIT_PAY => $this->waitPaySubStock(),
+                OrderCode::UNCOMPLETED_REFUND => $this->unFinishRefundAddStock(),
+                OrderCode::CANCEL => $this->cancelAddStock(),
+                OrderCode::FINISH_REFUND => $this->finishRefundUpdateData(),
+                default => throw new Exception('OrderGoodStockConsumer:error:无效的订单类型')
+            };
+
+            if (empty($this->updateArr)) {
+                $this->log->debug('OrderGoodStockConsumer:error:NoUpdateData:skuInfo:'.json_encode($this->skuArr).':orderGoodArr:'.json_encode($this->orderGoodArr));
+                return Result::ACK;
+            }
+
+            if (!(new Sku)->update($this->updateArr)) {
+                $this->log->debug('OrderGoodStockConsumer:error:UpdateSkuDataFail:'.json_encode($this->updateArr));
+                return Result::ACK;
+            }
+
+            return Result::ACK;
+        } catch (Exception $e) {
+            $this->log->error($e->getMessage());
+            return Result::ACK;
+        }
+    }
+
+    /**
+     * @return void
+     */
+    private function waitPaySubStock(): void
+    {
+        foreach ($this->orderGoodArr as $orderGood) {
+            $this->updateArr[] = [
+                'id' => $orderGood['sku_id'],
+                ''
+            ];
+        }
+    }
+
+    /**
+     * @return void
+     */
+    private function unFinishRefundAddStock(): void
+    {
+        foreach ($this->orderGoodArr as $orderGood) {
+            $this->updateArr[] = [
+                'id' => $orderGood['sku_id'],
+                ''
+            ];
+        }
+    }
+
+    /**
+     * @return void
+     */
+    private function cancelAddStock(): void
+    {
+        foreach ($this->orderGoodArr as $orderGood) {
+            $this->updateArr[] = [
+                'id' => $orderGood['sku_id'],
+                ''
+            ];
+        }
+    }
+
+    /**
+     * @return void
+     */
+    private function finishRefundUpdateData(): void
+    {
+        foreach ($this->orderGoodArr as $orderGood) {
+            $this->updateArr[] = [
+                'id' => $orderGood['sku_id'],
+                ''
+            ];
+        }
     }
 }

app/Model/OrderGood.php

@@ -42,4 +42,14 @@ class OrderGood extends Model
     const string CREATED_AT = 'create_time';
 
     const string UPDATED_AT = 'update_time';
+
+    /**
+     * 根据订单获取商品id
+     * @param int $orderId
+     * @return array
+     */
+    public function getGoodIdsByOrderId(int $orderId): array
+    {
+        return $this->where('order_id', $orderId)->select('sku_id','SUM(`quantity`) as `quantity`')->get()->toArray();
+    }
 }

app/Model/Sku.php

@@ -75,4 +75,20 @@ class Sku extends Model
             ->select(['id','spu_id','title','image_ids','price','param','extra','total_stock','surplus_stock','order_num'])
             ->get();
     }
+
+    /**
+     * @param array $ids
+     * @return array
+     */
+    public function getDataArrByIds(array $ids): array
+    {
+        $res = $this
+            ->whereIn('id',$ids)
+            ->orderBy('sort')
+            ->get();
+
+        if (empty($res)) return [];
+
+        return $res->toArray();
+    }
 }

app/Service/Api/Order/PlaceOrderService.php

@@ -33,6 +33,11 @@ class PlaceOrderService extends BaseOrderService
         $this->isAutoSelectCoupon = 2;
     }
 
+    /**
+     * @var array
+     */
+    private array $rollbackStockCache = [];
+
 
     /**
      * 统一下单逻辑
@@ -45,12 +50,16 @@ class PlaceOrderService extends BaseOrderService
         $this->siteId = (int)$this->request->input('site_id');
         $this->couponId = (int)$this->request->input('coupon_id',0);
 
+        // 生成购物车信息 ， 检测商品和库存等
         $this->check();
 
+        // 计算
         $this->compute();
 
+        // 下单 减少库存 写入数据库
         $this->placeOrder();
 
+        // 加入取消延迟队列
         $this->joinCancelDelayQueue();
 
         return $this->return->success('success',$this->orderRes);
@@ -73,22 +82,61 @@ class PlaceOrderService extends BaseOrderService
         $producer->produce($message);
     }
 
+    /**
+     * @return void
+     * @throws ContainerExceptionInterface
+     * @throws NotFoundExceptionInterface
+     * @throws Exception
+     */
+    private function reduceStock(): void
+    {
+        $this->rollbackStockCache = [];
+
+        foreach ($this->cartFirstData as $goodId => $stock) {
+            $this->rollbackStockCache[$goodId] = $stock;
+            if (!($this->redis->zAdd($this->stockKey,'-'.$stock,$goodId))) {
+                throw new Exception('cache error');
+            }
+        }
+    }
+
+    /**
+     * @return void
+     * @throws ContainerExceptionInterface
+     * @throws NotFoundExceptionInterface
+     */
+    private function rollbackStock(): void
+    {
+        if (empty($this->rollbackStockCache)) return;
+
+        foreach ($this->rollbackStockCache as $goodId => $stock) {
+            $this->redis->zAdd($this->stockKey,$stock,$goodId);
+        }
+    }
+
     /**
      * 下单
      * @return void
+     * @throws ContainerExceptionInterface
+     * @throws NotFoundExceptionInterface
      */
     private function placeOrder(): void
     {
         try {
             Db::beginTransaction();
 
+            $this->reduceStock();
+
             $this->insertOrder();
 
             $this->insertOrderGoods();
 
             Db::commit();
         } catch (Exception $e){
+            //回滚数据库 和 缓存
             Db::rollBack();
+            $this->rollbackStock();
+            //意外抛出
             throw new ErrException($e->getMessage());
         }
     }

app/Service/ServiceTrait/Api/OrderTrait.php

@@ -10,6 +10,7 @@ declare(strict_types=1);
 
 namespace App\Service\ServiceTrait\Api;
 
+use App\Amqp\Producer\OrderGoodStockProducer;
 use App\Cache\Redis\Api\ApiRedisKey;
 use App\Cache\Redis\Api\SiteCache;
 use App\Cache\Redis\RedisCache;
@@ -17,6 +18,8 @@ use App\Constants\ApiCode;
 use App\Constants\Common\GoodCode;
 use App\Constants\ConfigCode;
 use App\Exception\ErrException;
+use Hyperf\Amqp\Producer;
+use Hyperf\Context\ApplicationContext;
 use Hyperf\Di\Annotation\Inject;
 use Psr\Container\ContainerExceptionInterface;
 use Psr\Container\NotFoundExceptionInterface;
@@ -260,4 +263,21 @@ trait OrderTrait
         $this->orderRes['total_price'] = bcadd($this->orderRes['total_good_price'],$this->orderRes['total_sundry_price'],2);
         $this->orderRes['actual_price'] = bcadd($this->orderRes['good_after_discount_price'],$this->orderRes['sundry_after_discount_price'],2);
     }
+
+    /**
+     * @param int $orderId
+     * @param int $orderStatus
+     * @return void
+     * @throws ContainerExceptionInterface
+     * @throws NotFoundExceptionInterface
+     */
+    protected function sendStockMq(int $orderId,int $orderStatus): void
+    {
+        $message = new OrderGoodStockProducer([
+            'order_id' => $orderId,
+            'type' => $orderStatus
+        ]);
+        $producer = ApplicationContext::getContainer()->get(Producer::class);
+        $producer->produce($message);
+    }
 }

app/Service/ServiceTrait/Common/StockTrait.php

@@ -0,0 +1,8 @@
+<?php
+
+namespace App\Service\ServiceTrait\Common;
+
+trait StockTrait
+{
+
+}