330 lines
10 KiB
Go
330 lines
10 KiB
Go
|
// 库存管理服务（Stock公共库存）
|
|||
|
|
// 职责：入库/出库操作，支持明细模式(StockDetails)和批次模式(StockBatch)
|
|||
|
|
// 调用链：Controller → StockOperation → stockPublishMessage → NATS → AddStock(消费者)
|
|||
|
|
// 紧密耦合：dao.StockDetails、dao.StockBatch、dao.AssetSku(更新库存数)、common/message(NATS发布)
|
|||
|
|
// 注意：移库/调拨是PrivateStock专属操作，不在此实现
|
|||
|
|
package service
|
|||
|
|
|
|||
|
|
import (
|
|||
|
|
"assets/consts/public"
|
|||
|
|
"assets/consts/stock"
|
|||
|
|
assetDao "assets/dao/asset"
|
|||
|
|
dao "assets/dao/stock"
|
|||
|
|
assetDto "assets/model/dto/asset"
|
|||
|
|
stockDto "assets/model/dto/stock"
|
|||
|
|
assetEntity "assets/model/entity/asset"
|
|||
|
|
entity "assets/model/entity/stock"
|
|||
|
|
"context"
|
|||
|
|
"fmt"
|
|||
|
|
|
|||
|
|
"gitea.com/red-future/common/beans"
|
|||
|
|
"gitea.com/red-future/common/redis"
|
|||
|
|
"gitea.com/red-future/common/utils"
|
|||
|
|
"github.com/gogf/gf/v2/errors/gerror"
|
|||
|
|
"github.com/gogf/gf/v2/frame/g"
|
|||
|
|
"github.com/gogf/gf/v2/os/gtime"
|
|||
|
|
"github.com/gogf/gf/v2/util/gconv"
|
|||
|
|
"go.mongodb.org/mongo-driver/v2/bson"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
type stockManage struct{}
|
|||
|
|
|
|||
|
|
// StockManage 库存管理服务（Stock公共库存）
|
|||
|
|
// 职责：
|
|||
|
|
// 1. 入库/出库：StockDetails（明细模式）和 StockBatch（批次模式）的库存操作
|
|||
|
|
// 2. 库存查询表单字段生成
|
|||
|
|
// 注意：移库（MoveStock）和调拨（TransferStock）是PrivateStock专属操作，不要在这里实现
|
|||
|
|
var StockManage = new(stockManage)
|
|||
|
|
|
|||
|
|
// GetStockFormFields 获取库存操作表单字段
|
|||
|
|
func (s *stockManage) GetStockFormFields(ctx context.Context, req *stockDto.GetStockFormFieldsReq) (*stockDto.GetStockFormFieldsRes, error) {
|
|||
|
|
// 获取资产SKU信息
|
|||
|
|
assetSku, err := assetDao.AssetSku.GetOne(ctx, &assetDto.GetAssetSkuReq{Id: req.AssetSkuId}, false)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
fields := make([]map[string]interface{}, 0)
|
|||
|
|
// Stock 字段在两种模式下都显示
|
|||
|
|
fields = append(fields, map[string]interface{}{
|
|||
|
|
"name": "stock",
|
|||
|
|
"label": "库存数量",
|
|||
|
|
"type": "number",
|
|||
|
|
"required": true,
|
|||
|
|
"min": 1,
|
|||
|
|
})
|
|||
|
|
// 如果是批次模式(2)，添加批次相关字段
|
|||
|
|
if !g.IsEmpty(assetSku.StockMode) && assetSku.StockMode == stock.StockModeBatch {
|
|||
|
|
fields = append(fields, []map[string]interface{}{
|
|||
|
|
{
|
|||
|
|
"name": "batchNo",
|
|||
|
|
"label": "批次号",
|
|||
|
|
"type": "number",
|
|||
|
|
"required": true,
|
|||
|
|
"default": gconv.Int(gtime.Now().Format("20060102") + "0001"),
|
|||
|
|
"maxLength": 12,
|
|||
|
|
},
|
|||
|
|
{
|
|||
|
|
"name": "productionDate",
|
|||
|
|
"label": "生产日期",
|
|||
|
|
"type": "date",
|
|||
|
|
},
|
|||
|
|
{
|
|||
|
|
"name": "expiryDate",
|
|||
|
|
"label": "过期日期",
|
|||
|
|
"type": "date",
|
|||
|
|
},
|
|||
|
|
{
|
|||
|
|
"name": "expiryWarningDate",
|
|||
|
|
"label": "临期预警时间",
|
|||
|
|
"type": "date",
|
|||
|
|
},
|
|||
|
|
}...)
|
|||
|
|
}
|
|||
|
|
return &stockDto.GetStockFormFieldsRes{
|
|||
|
|
StockMode: assetSku.StockMode,
|
|||
|
|
Fields: fields,
|
|||
|
|
}, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// StockOperation 库存操作入口（入库/出库）
|
|||
|
|
// 根据SKU的StockMode区分明细模式和批次模式，计算差值后发布消息到NATS
|
|||
|
|
func (s *stockManage) StockOperation(ctx context.Context, req *stockDto.StockOperationReq) (err error) {
|
|||
|
|
assetSku, err := assetDao.AssetSku.GetOne(ctx, &assetDto.GetAssetSkuReq{Id: req.AssetSkuId}, false)
|
|||
|
|
if err != nil {
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
if !assetSku.UnlimitedStock && req.Stock >= 0 {
|
|||
|
|
var stockId *bson.ObjectID
|
|||
|
|
count := 0
|
|||
|
|
if assetSku.StockMode == stock.StockModeDetail {
|
|||
|
|
_count, err := dao.StockDetails.GetStockCountBySkuId(ctx, assetSku.Id)
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
count = gconv.Int(_count)
|
|||
|
|
}
|
|||
|
|
if assetSku.StockMode == stock.StockModeBatch {
|
|||
|
|
if g.IsEmpty(req.BatchNo) {
|
|||
|
|
return gerror.New("批次号不能为空")
|
|||
|
|
}
|
|||
|
|
getOne, err := dao.StockBatch.GetOne(ctx, req.BatchNo)
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
if !g.IsEmpty(getOne) {
|
|||
|
|
stockId = getOne.Id
|
|||
|
|
count = getOne.BatchQty
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
stockCount := 0
|
|||
|
|
operationType := ""
|
|||
|
|
if count != req.Stock {
|
|||
|
|
if count > req.Stock {
|
|||
|
|
stockCount = count - req.Stock
|
|||
|
|
operationType = "del"
|
|||
|
|
} else {
|
|||
|
|
stockCount = req.Stock - count
|
|||
|
|
operationType = "add"
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
if !g.IsEmpty(operationType) && stockCount > 0 {
|
|||
|
|
if err = s.stockPublishMessage(ctx, assetSku, stockId, stockCount, operationType, req); err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// stockPublishMessage 发布库存变更消息到NATS
|
|||
|
|
// 消费者接收后执行实际的入库/出库操作（异步解耦）
|
|||
|
|
func (s *stockManage) stockPublishMessage(ctx context.Context, assetSku *assetEntity.AssetSku, stockId *bson.ObjectID, stockCount int, operationType string, req *stockDto.StockOperationReq) (err error) {
|
|||
|
|
// 用户信息
|
|||
|
|
user, err := utils.GetUserInfo(ctx)
|
|||
|
|
if err != nil {
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
publishMessage := stockDto.StockPublishMessage{
|
|||
|
|
AssetId: assetSku.AssetId.Hex(),
|
|||
|
|
AssetSkuId: assetSku.Id.Hex(),
|
|||
|
|
TenantId: user.TenantId,
|
|||
|
|
UserName: user.UserName,
|
|||
|
|
StockCount: stockCount,
|
|||
|
|
OperationType: operationType,
|
|||
|
|
Metadata: assetSku.SpecValues,
|
|||
|
|
StockMode: int(assetSku.StockMode),
|
|||
|
|
BatchNo: req.BatchNo,
|
|||
|
|
ProductionDate: req.ProductionDate,
|
|||
|
|
ExpiryDate: req.ExpiryDate,
|
|||
|
|
ExpiryWarningDate: req.ExpiryWarningDate,
|
|||
|
|
}
|
|||
|
|
if !g.IsEmpty(stockId) && !stockId.IsZero() {
|
|||
|
|
publishMessage.StockId = stockId.Hex()
|
|||
|
|
}
|
|||
|
|
// 发布到 NATS
|
|||
|
|
//plugin, err := message.GetMsgPlugin(ctx, message.MessageNATS)
|
|||
|
|
//if err != nil {
|
|||
|
|
// return gerror.Newf("NATS插件未就绪: %v", err)
|
|||
|
|
//}
|
|||
|
|
//err = plugin.Publish(ctx, &message.NatsPublishMsgConfig{
|
|||
|
|
// QueueName: public.StockDetailGroupName,
|
|||
|
|
// Durable: true,
|
|||
|
|
// Data: publishMessage,
|
|||
|
|
//})
|
|||
|
|
//_, err = message.PublishMessage(ctx, &message.RedisMessageConfig{StreamKey: public.StockDetailStreamKey}, publishMessage)
|
|||
|
|
//plugin, err := message.GetMsgPlugin(message.MessageRedis)
|
|||
|
|
//if err != nil {
|
|||
|
|
// return err
|
|||
|
|
//}
|
|||
|
|
//err = plugin.Publish(ctx, &message.RedisPublishMsgConfig{
|
|||
|
|
// QueueName: public.StockDetailQueueName,
|
|||
|
|
// Data: publishMessage,
|
|||
|
|
//})
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// AddStock NATS消费者调用，执行实际的入库/出库操作
|
|||
|
|
// 使用Redis分布式锁防止并发冲突，支持明细模式和批次模式
|
|||
|
|
func (s *stockManage) AddStock(ctx context.Context, msg map[string]interface{}) error {
|
|||
|
|
assetId, err := bson.ObjectIDFromHex(gconv.String(msg["assetId"]))
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
assetSkuId, err := bson.ObjectIDFromHex(gconv.String(msg["assetSkuId"]))
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
stockId := bson.ObjectID{}
|
|||
|
|
if !g.IsEmpty(msg["stockId"]) {
|
|||
|
|
stockId, err = bson.ObjectIDFromHex(gconv.String(msg["stockId"]))
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
userName := gconv.String(msg["userName"])
|
|||
|
|
tenantId := gconv.Float64(msg["tenantId"])
|
|||
|
|
stockCount := gconv.Int(msg["stockCount"])
|
|||
|
|
operationType := gconv.String(msg["operationType"])
|
|||
|
|
metadata := gconv.Maps(msg["metadata"])
|
|||
|
|
stockMode := stock.StockMode(gconv.Int(msg["stockMode"]))
|
|||
|
|
batchNo := gconv.String(msg["batchNo"])
|
|||
|
|
productionDate := gtime.New(msg["productionDate"])
|
|||
|
|
expiryDate := gtime.New(msg["expiryDate"])
|
|||
|
|
expiryWarningDate := gtime.New(msg["expiryWarningDate"])
|
|||
|
|
// 设置 userId 和 tenantId 到 ctx
|
|||
|
|
ctx = context.WithValue(ctx, "userName", userName)
|
|||
|
|
ctx = context.WithValue(ctx, "tenantId", tenantId)
|
|||
|
|
// 获取redis-租户存储-锁key
|
|||
|
|
fileLockKey := fmt.Sprintf(public.StockDetailLockKey, assetSkuId)
|
|||
|
|
success, err := redis.Lock(ctx, fileLockKey, int64(60), func(ctx context.Context) error {
|
|||
|
|
if operationType == "add" {
|
|||
|
|
if stockMode == stock.StockModeBatch {
|
|||
|
|
if !stockId.IsZero() {
|
|||
|
|
batch := stockDto.UpdateBatchReq{
|
|||
|
|
Id: &stockId,
|
|||
|
|
BatchQty: stockCount,
|
|||
|
|
AvailableQty: stockCount,
|
|||
|
|
}
|
|||
|
|
if err := dao.StockBatch.Update(ctx, &batch); err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
} else {
|
|||
|
|
batch := stockDto.CreateBatchReq{
|
|||
|
|
AssetId: &assetId,
|
|||
|
|
AssetSkuId: &assetSkuId,
|
|||
|
|
Status: stock.BatchStatusActive,
|
|||
|
|
Metadata: metadata,
|
|||
|
|
BatchNo: batchNo,
|
|||
|
|
BatchQty: stockCount,
|
|||
|
|
AvailableQty: stockCount,
|
|||
|
|
ProductionDate: productionDate,
|
|||
|
|
ExpiryDate: expiryDate,
|
|||
|
|
ExpiryWarningDate: expiryWarningDate,
|
|||
|
|
}
|
|||
|
|
if _, err := dao.StockBatch.Insert(ctx, &batch); err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
if stockMode == stock.StockModeDetail {
|
|||
|
|
// 创建指定数量的库存
|
|||
|
|
var stockInterfaces []interface{}
|
|||
|
|
for i := 0; i < stockCount; i++ {
|
|||
|
|
stockInterfaces = append(stockInterfaces, entity.StockDetails{
|
|||
|
|
AssetId: &assetId,
|
|||
|
|
AssetSkuId: &assetSkuId,
|
|||
|
|
Status: stock.StockStatusAvailable,
|
|||
|
|
Metadata: metadata,
|
|||
|
|
})
|
|||
|
|
}
|
|||
|
|
// 批量插入数据库
|
|||
|
|
if _, err = dao.StockDetails.BatchInsert(ctx, stockInterfaces); err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
if operationType == "del" {
|
|||
|
|
if stockMode == stock.StockModeBatch {
|
|||
|
|
stockCount = 0 - stockCount
|
|||
|
|
// 更新批次
|
|||
|
|
batch := stockDto.UpdateBatchReq{
|
|||
|
|
Id: &stockId,
|
|||
|
|
BatchQty: stockCount,
|
|||
|
|
AvailableQty: stockCount,
|
|||
|
|
}
|
|||
|
|
if err := dao.StockBatch.Update(ctx, &batch); err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
if stockMode == stock.StockModeDetail {
|
|||
|
|
// 分页查询所有库存明细，收集所有ID
|
|||
|
|
var allStockIds []*bson.ObjectID
|
|||
|
|
pageSize := int64(50)
|
|||
|
|
for pageNum := int64(1); ; pageNum++ {
|
|||
|
|
details, total, err := dao.StockDetails.List(ctx,
|
|||
|
|
&stockDto.ListStockDetailsReq{
|
|||
|
|
AssetSkuId: &assetSkuId,
|
|||
|
|
Status: stock.StockStatusAvailable,
|
|||
|
|
Page: &beans.Page{PageNum: pageNum, PageSize: pageSize},
|
|||
|
|
})
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
if pageNum == 1 && int(total) < stockCount {
|
|||
|
|
return gerror.New("可操作库存数量不足")
|
|||
|
|
}
|
|||
|
|
// 收集当前页的ID
|
|||
|
|
for _, detail := range details {
|
|||
|
|
if detail.Id != nil && !detail.Id.IsZero() {
|
|||
|
|
allStockIds = append(allStockIds, detail.Id)
|
|||
|
|
if len(allStockIds) >= stockCount {
|
|||
|
|
break
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
if len(allStockIds) >= stockCount {
|
|||
|
|
break
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
// 根据ID批量删除库存
|
|||
|
|
delCount, err := dao.StockDetails.DeleteManyByIds(ctx, allStockIds)
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
if delCount != int64(stockCount) {
|
|||
|
|
return gerror.New("删除库存数量不匹配")
|
|||
|
|
}
|
|||
|
|
stockCount = 0 - stockCount
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
return assetDao.AssetSku.Update(ctx, &assetDto.UpdateAssetSkuReq{Id: &assetSkuId, Stock: stockCount})
|
|||
|
|
})
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
if !success {
|
|||
|
|
return fmt.Errorf("获取库存操作锁失败: %v", err)
|
|||
|
|
}
|
|||
|
|
return nil
|
|||
|
|
}
|