实现资产双向换货链路查询
This commit is contained in:
228
internal/query/asset/exchange_trace.go
Normal file
228
internal/query/asset/exchange_trace.go
Normal file
@@ -0,0 +1,228 @@
|
||||
// Package asset 提供资产详情读取投影。
|
||||
package asset
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||||
"github.com/break/junhong_cmp_fiber/internal/model/dto"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/middleware"
|
||||
"go.uber.org/zap"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// ExchangeTraceQuery 查询资产单节点换货链路并投影关联资产可见性。
|
||||
type ExchangeTraceQuery struct {
|
||||
db *gorm.DB
|
||||
logger *zap.Logger
|
||||
}
|
||||
|
||||
// ExchangeTraceAssetRef 标识批量换货关系查询中的当前资产。
|
||||
type ExchangeTraceAssetRef struct {
|
||||
// AssetType 为统一资产详情或换货模型中的资产类型。
|
||||
AssetType string
|
||||
// AssetID 为资产数据库 ID。
|
||||
AssetID uint
|
||||
}
|
||||
|
||||
// NewExchangeTraceQuery 创建资产换货链路查询。
|
||||
func NewExchangeTraceQuery(db *gorm.DB, logger *zap.Logger) *ExchangeTraceQuery {
|
||||
if logger == nil {
|
||||
logger = zap.NewNop()
|
||||
}
|
||||
return &ExchangeTraceQuery{db: db, logger: logger}
|
||||
}
|
||||
|
||||
// Resolve 查询当前资产的前代与后代换货投影。
|
||||
func (q *ExchangeTraceQuery) Resolve(ctx context.Context, assetType string, assetID uint) (*dto.AssetExchangeTrace, error) {
|
||||
trace := &dto.AssetExchangeTrace{}
|
||||
assetType = normalizeExchangeAssetType(assetType)
|
||||
ref := ExchangeTraceAssetRef{AssetType: assetType, AssetID: assetID}
|
||||
previousBatch, err := q.FindPreviousCompleted(ctx, []ExchangeTraceAssetRef{ref})
|
||||
if err != nil {
|
||||
return nil, q.wrapQueryError(constants.ExchangeTraceDirectionPrevious, assetType, assetID, err)
|
||||
}
|
||||
nextBatch, err := q.FindNextCompleted(ctx, []ExchangeTraceAssetRef{ref})
|
||||
if err != nil {
|
||||
return nil, q.wrapQueryError(constants.ExchangeTraceDirectionNext, assetType, assetID, err)
|
||||
}
|
||||
|
||||
previousOrders := previousBatch[ref]
|
||||
nextOrders := nextBatch[ref]
|
||||
previous := q.selectLatest(constants.ExchangeTraceDirectionPrevious, assetType, assetID, previousOrders)
|
||||
next := q.selectLatest(constants.ExchangeTraceDirectionNext, assetType, assetID, nextOrders)
|
||||
visibility, err := q.loadVisibility(ctx, previous, next)
|
||||
if err != nil {
|
||||
q.logger.Error("查询换货关联资产可见性失败",
|
||||
zap.String("asset_type", assetType),
|
||||
zap.Uint("asset_id", assetID),
|
||||
zap.Error(err))
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询换货关联资产可见性失败")
|
||||
}
|
||||
|
||||
if previous != nil {
|
||||
key := relatedAssetKey{assetType: previous.OldAssetType, assetID: previous.OldAssetID}
|
||||
previousAssetID := previous.OldAssetID
|
||||
trace.PreviousAsset = newTraceItem(previous.OldAssetType, &previousAssetID, previous.OldAssetIdentifier, previous.ExchangeNo, visibility[key])
|
||||
}
|
||||
if next != nil {
|
||||
var canView bool
|
||||
if next.NewAssetID != nil {
|
||||
key := relatedAssetKey{assetType: next.NewAssetType, assetID: *next.NewAssetID}
|
||||
canView = visibility[key]
|
||||
} else {
|
||||
q.logger.Warn("已完成换货记录缺少新资产 ID",
|
||||
zap.String("asset_type", assetType),
|
||||
zap.Uint("asset_id", assetID),
|
||||
zap.Uint("exchange_id", next.ID),
|
||||
zap.String("exchange_no", next.ExchangeNo))
|
||||
}
|
||||
trace.NextAsset = newTraceItem(next.NewAssetType, next.NewAssetID, next.NewAssetIdentifier, next.ExchangeNo, canView)
|
||||
}
|
||||
return trace, nil
|
||||
}
|
||||
|
||||
func normalizeExchangeAssetType(assetType string) string {
|
||||
if assetType == constants.AssetResolveTypeCard {
|
||||
return constants.ExchangeAssetTypeIotCard
|
||||
}
|
||||
return assetType
|
||||
}
|
||||
|
||||
type relatedAssetKey struct {
|
||||
assetType string
|
||||
assetID uint
|
||||
}
|
||||
|
||||
// FindPreviousCompleted 批量查询当前资产作为新资产时的已完成换货记录。
|
||||
func (q *ExchangeTraceQuery) FindPreviousCompleted(ctx context.Context, refs []ExchangeTraceAssetRef) (map[ExchangeTraceAssetRef][]*model.ExchangeOrder, error) {
|
||||
return q.findRelatedOrdersBatch(ctx, constants.ExchangeTraceDirectionPrevious, refs)
|
||||
}
|
||||
|
||||
// FindNextCompleted 批量查询当前资产作为旧资产时的已完成换货记录。
|
||||
func (q *ExchangeTraceQuery) FindNextCompleted(ctx context.Context, refs []ExchangeTraceAssetRef) (map[ExchangeTraceAssetRef][]*model.ExchangeOrder, error) {
|
||||
return q.findRelatedOrdersBatch(ctx, constants.ExchangeTraceDirectionNext, refs)
|
||||
}
|
||||
|
||||
func (q *ExchangeTraceQuery) findRelatedOrdersBatch(ctx context.Context, direction string, refs []ExchangeTraceAssetRef) (map[ExchangeTraceAssetRef][]*model.ExchangeOrder, error) {
|
||||
result := make(map[ExchangeTraceAssetRef][]*model.ExchangeOrder, len(refs))
|
||||
if len(refs) == 0 {
|
||||
return result, nil
|
||||
}
|
||||
var orders []*model.ExchangeOrder
|
||||
query := q.db.WithContext(ctx).Where("status = ?", constants.ExchangeStatusCompleted)
|
||||
assetConditions := q.db.Session(&gorm.Session{NewDB: true}).Where("1 = 0")
|
||||
for _, ref := range refs {
|
||||
ref.AssetType = normalizeExchangeAssetType(ref.AssetType)
|
||||
if direction == constants.ExchangeTraceDirectionPrevious {
|
||||
assetConditions = assetConditions.Or("new_asset_type = ? AND new_asset_id = ?", ref.AssetType, ref.AssetID)
|
||||
} else {
|
||||
assetConditions = assetConditions.Or("old_asset_type = ? AND old_asset_id = ?", ref.AssetType, ref.AssetID)
|
||||
}
|
||||
}
|
||||
err := query.Where(assetConditions).
|
||||
Order("completed_at DESC NULLS LAST, id DESC").
|
||||
Find(&orders).Error
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, order := range orders {
|
||||
ref := ExchangeTraceAssetRef{AssetType: order.OldAssetType, AssetID: order.OldAssetID}
|
||||
if direction == constants.ExchangeTraceDirectionPrevious && order.NewAssetID != nil {
|
||||
ref = ExchangeTraceAssetRef{AssetType: order.NewAssetType, AssetID: *order.NewAssetID}
|
||||
}
|
||||
result[ref] = append(result[ref], order)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (q *ExchangeTraceQuery) selectLatest(direction string, assetType string, assetID uint, orders []*model.ExchangeOrder) *model.ExchangeOrder {
|
||||
if len(orders) == 0 {
|
||||
return nil
|
||||
}
|
||||
selected := orders[0]
|
||||
if len(orders) > 1 {
|
||||
q.logger.Warn("检测到同方向多条已完成换货记录",
|
||||
zap.String("direction", string(direction)),
|
||||
zap.String("asset_type", assetType),
|
||||
zap.Uint("asset_id", assetID),
|
||||
zap.Int("candidate_count", len(orders)),
|
||||
zap.Uint("selected_exchange_id", selected.ID),
|
||||
zap.String("selected_exchange_no", selected.ExchangeNo))
|
||||
}
|
||||
return selected
|
||||
}
|
||||
|
||||
func (q *ExchangeTraceQuery) loadVisibility(ctx context.Context, previous, next *model.ExchangeOrder) (map[relatedAssetKey]bool, error) {
|
||||
result := make(map[relatedAssetKey]bool, 2)
|
||||
cardIDs, deviceIDs := relatedAssetIDs(previous, next)
|
||||
if len(cardIDs) > 0 {
|
||||
var cards []*model.IotCard
|
||||
if err := q.db.WithContext(ctx).Select("id", "shop_id").Where("id IN ?", cardIDs).Find(&cards).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, card := range cards {
|
||||
result[relatedAssetKey{assetType: constants.ExchangeAssetTypeIotCard, assetID: card.ID}] = canViewShopAsset(ctx, card.ShopID)
|
||||
}
|
||||
}
|
||||
if len(deviceIDs) > 0 {
|
||||
var devices []*model.Device
|
||||
if err := q.db.WithContext(ctx).Select("id", "shop_id").Where("id IN ?", deviceIDs).Find(&devices).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, device := range devices {
|
||||
result[relatedAssetKey{assetType: constants.ExchangeAssetTypeDevice, assetID: device.ID}] = canViewShopAsset(ctx, device.ShopID)
|
||||
}
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func canViewShopAsset(ctx context.Context, shopID *uint) bool {
|
||||
if middleware.IsUnrestricted(ctx) {
|
||||
return true
|
||||
}
|
||||
return shopID != nil && middleware.ContainsShopID(ctx, *shopID)
|
||||
}
|
||||
|
||||
func relatedAssetIDs(previous, next *model.ExchangeOrder) ([]uint, []uint) {
|
||||
cardIDs := make([]uint, 0, 2)
|
||||
deviceIDs := make([]uint, 0, 2)
|
||||
appendID := func(assetType string, assetID uint) {
|
||||
if assetType == constants.ExchangeAssetTypeIotCard {
|
||||
cardIDs = append(cardIDs, assetID)
|
||||
} else if assetType == constants.ExchangeAssetTypeDevice {
|
||||
deviceIDs = append(deviceIDs, assetID)
|
||||
}
|
||||
}
|
||||
if previous != nil {
|
||||
appendID(previous.OldAssetType, previous.OldAssetID)
|
||||
}
|
||||
if next != nil && next.NewAssetID != nil {
|
||||
appendID(next.NewAssetType, *next.NewAssetID)
|
||||
}
|
||||
return cardIDs, deviceIDs
|
||||
}
|
||||
|
||||
func (q *ExchangeTraceQuery) wrapQueryError(direction string, assetType string, assetID uint, err error) error {
|
||||
q.logger.Error("查询资产换货关系失败",
|
||||
zap.String("direction", string(direction)),
|
||||
zap.String("asset_type", assetType),
|
||||
zap.Uint("asset_id", assetID),
|
||||
zap.Error(err))
|
||||
return errors.Wrap(errors.CodeDatabaseError, err, "查询资产换货关系失败")
|
||||
}
|
||||
|
||||
func newTraceItem(assetType string, assetID *uint, identifier, exchangeNo string, canView bool) *dto.AssetExchangeTraceItem {
|
||||
item := &dto.AssetExchangeTraceItem{
|
||||
AssetType: assetType,
|
||||
Identifier: identifier,
|
||||
ExchangeNo: exchangeNo,
|
||||
CanView: canView,
|
||||
}
|
||||
if canView && assetID != nil {
|
||||
item.AssetID = assetID
|
||||
}
|
||||
return item
|
||||
}
|
||||
435
internal/query/asset/exchange_trace_test.go
Normal file
435
internal/query/asset/exchange_trace_test.go
Normal file
@@ -0,0 +1,435 @@
|
||||
package asset
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||||
"github.com/break/junhong_cmp_fiber/internal/model/dto"
|
||||
"github.com/break/junhong_cmp_fiber/internal/testutil"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||||
"go.uber.org/zap"
|
||||
"go.uber.org/zap/zaptest/observer"
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/logger"
|
||||
)
|
||||
|
||||
// TestExchangeTraceQueryProjectsPreviousAsset 验证卡和设备前代均使用换货快照，并始终返回稳定对象。
|
||||
func TestExchangeTraceQueryProjectsPreviousAsset(t *testing.T) {
|
||||
tx := testutil.NewPostgresTransaction(t)
|
||||
query := NewExchangeTraceQuery(tx, zap.NewNop())
|
||||
|
||||
card := createTraceCard(t, tx, 1, nil)
|
||||
trace, err := query.Resolve(context.Background(), constants.ExchangeAssetTypeIotCard, card.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("查询无换货关系资产失败：%v", err)
|
||||
}
|
||||
if trace == nil || trace.PreviousAsset != nil || trace.NextAsset != nil {
|
||||
t.Fatalf("无关系时应返回稳定空对象：%+v", trace)
|
||||
}
|
||||
|
||||
oldCard := createTraceCard(t, tx, 2, nil)
|
||||
createTraceOrder(t, tx, "UR86-PREV-CARD", constants.ExchangeAssetTypeIotCard, oldCard.ID, "历史卡快照", constants.ExchangeAssetTypeIotCard, card.ID, "新卡快照", constants.ExchangeStatusCompleted, time.Now())
|
||||
trace, err = query.Resolve(context.Background(), constants.ExchangeAssetTypeIotCard, card.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("查询卡前代失败：%v", err)
|
||||
}
|
||||
assertTraceAsset(t, trace.PreviousAsset, constants.ExchangeAssetTypeIotCard, oldCard.ID, "历史卡快照", "UR86-PREV-CARD", true)
|
||||
|
||||
oldDevice := createTraceDevice(t, tx, 3, nil)
|
||||
newDevice := createTraceDevice(t, tx, 4, nil)
|
||||
createTraceOrder(t, tx, "UR86-PREV-DEVICE", constants.ExchangeAssetTypeDevice, oldDevice.ID, "历史设备快照", constants.ExchangeAssetTypeDevice, newDevice.ID, "新设备快照", constants.ExchangeStatusCompleted, time.Now())
|
||||
trace, err = query.Resolve(context.Background(), constants.ExchangeAssetTypeDevice, newDevice.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("查询设备前代失败：%v", err)
|
||||
}
|
||||
assertTraceAsset(t, trace.PreviousAsset, constants.ExchangeAssetTypeDevice, oldDevice.ID, "历史设备快照", "UR86-PREV-DEVICE", true)
|
||||
}
|
||||
|
||||
// TestExchangeTraceQueryHidesInvisiblePreviousAsset 验证关系查询不受权限过滤，但关联资产 ID 按当前权限隐藏。
|
||||
func TestExchangeTraceQueryHidesInvisiblePreviousAsset(t *testing.T) {
|
||||
tx := testutil.NewPostgresTransaction(t)
|
||||
visibleShop, hiddenShop := uint(86101), uint(86102)
|
||||
oldCard := createTraceCard(t, tx, 11, &hiddenShop)
|
||||
newCard := createTraceCard(t, tx, 12, &visibleShop)
|
||||
createTraceOrder(t, tx, "UR86-PREV-HIDDEN", constants.ExchangeAssetTypeIotCard, oldCard.ID, "不可见历史快照", constants.ExchangeAssetTypeIotCard, newCard.ID, "当前资产快照", constants.ExchangeStatusCompleted, time.Now())
|
||||
|
||||
ctx := context.WithValue(context.Background(), constants.ContextKeySubordinateShopIDs, []uint{visibleShop})
|
||||
trace, err := NewExchangeTraceQuery(tx, zap.NewNop()).Resolve(ctx, constants.ExchangeAssetTypeIotCard, newCard.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("查询不可见前代失败：%v", err)
|
||||
}
|
||||
assertTraceAsset(t, trace.PreviousAsset, constants.ExchangeAssetTypeIotCard, 0, "不可见历史快照", "UR86-PREV-HIDDEN", false)
|
||||
|
||||
emptyScope := context.WithValue(context.Background(), constants.ContextKeySubordinateShopIDs, []uint{})
|
||||
trace, err = NewExchangeTraceQuery(tx, zap.NewNop()).Resolve(emptyScope, constants.ExchangeAssetTypeIotCard, newCard.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("查询空权限范围前代失败：%v", err)
|
||||
}
|
||||
assertTraceAsset(t, trace.PreviousAsset, constants.ExchangeAssetTypeIotCard, 0, "不可见历史快照", "UR86-PREV-HIDDEN", false)
|
||||
}
|
||||
|
||||
// TestExchangeTraceQueryIgnoresIncompleteAndDeletedPreviousRelations 验证仅未软删除的已完成换货形成前代。
|
||||
func TestExchangeTraceQueryIgnoresIncompleteAndDeletedPreviousRelations(t *testing.T) {
|
||||
tx := testutil.NewPostgresTransaction(t)
|
||||
query := NewExchangeTraceQuery(tx, zap.NewNop())
|
||||
oldCard := createTraceCard(t, tx, 21, nil)
|
||||
statuses := []int{
|
||||
constants.ExchangeStatusPendingInfo,
|
||||
constants.ExchangeStatusPendingShip,
|
||||
constants.ExchangeStatusShipped,
|
||||
constants.ExchangeStatusCancelled,
|
||||
}
|
||||
for index, status := range statuses {
|
||||
newCard := createTraceCard(t, tx, 22+index, nil)
|
||||
createTraceOrder(t, tx, fmt.Sprintf("UR86-PREV-STATUS-%d", status), constants.ExchangeAssetTypeIotCard, oldCard.ID, "旧快照", constants.ExchangeAssetTypeIotCard, newCard.ID, "新快照", status, time.Now())
|
||||
trace, err := query.Resolve(context.Background(), constants.ExchangeAssetTypeIotCard, newCard.ID)
|
||||
if err != nil || trace.PreviousAsset != nil {
|
||||
t.Fatalf("状态 %d 不应形成前代：trace=%+v err=%v", status, trace, err)
|
||||
}
|
||||
}
|
||||
|
||||
deletedNew := createTraceCard(t, tx, 30, nil)
|
||||
deletedOrder := createTraceOrder(t, tx, "UR86-PREV-DELETED", constants.ExchangeAssetTypeIotCard, oldCard.ID, "旧快照", constants.ExchangeAssetTypeIotCard, deletedNew.ID, "新快照", constants.ExchangeStatusCompleted, time.Now())
|
||||
if err := tx.Delete(deletedOrder).Error; err != nil {
|
||||
t.Fatalf("软删除换货单失败：%v", err)
|
||||
}
|
||||
trace, err := query.Resolve(context.Background(), constants.ExchangeAssetTypeIotCard, deletedNew.ID)
|
||||
if err != nil || trace.PreviousAsset != nil {
|
||||
t.Fatalf("软删除换货单不应形成前代：trace=%+v err=%v", trace, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestExchangeTraceQuerySelectsLatestPreviousRelationAndLogsAnomaly 验证重复完成记录按完成时间和主键确定性选择。
|
||||
func TestExchangeTraceQuerySelectsLatestPreviousRelationAndLogsAnomaly(t *testing.T) {
|
||||
tx := testutil.NewPostgresTransaction(t)
|
||||
core, logs := observer.New(zap.WarnLevel)
|
||||
query := NewExchangeTraceQuery(tx, zap.New(core))
|
||||
newCard := createTraceCard(t, tx, 41, nil)
|
||||
oldOne := createTraceCard(t, tx, 42, nil)
|
||||
oldTwo := createTraceCard(t, tx, 43, nil)
|
||||
completedAt := time.Now().Truncate(time.Second)
|
||||
createTraceOrder(t, tx, "UR86-PREV-OLDER", constants.ExchangeAssetTypeIotCard, oldOne.ID, "较旧快照", constants.ExchangeAssetTypeIotCard, newCard.ID, "新快照", constants.ExchangeStatusCompleted, completedAt)
|
||||
createTraceOrder(t, tx, "UR86-PREV-LATEST", constants.ExchangeAssetTypeIotCard, oldTwo.ID, "最新快照", constants.ExchangeAssetTypeIotCard, newCard.ID, "新快照", constants.ExchangeStatusCompleted, completedAt)
|
||||
|
||||
trace, err := query.Resolve(context.Background(), constants.ExchangeAssetTypeIotCard, newCard.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("查询重复完成记录失败：%v", err)
|
||||
}
|
||||
assertTraceAsset(t, trace.PreviousAsset, constants.ExchangeAssetTypeIotCard, oldTwo.ID, "最新快照", "UR86-PREV-LATEST", true)
|
||||
if logs.FilterMessage("检测到同方向多条已完成换货记录").Len() != 1 {
|
||||
t.Fatalf("应记录重复换货异常日志：%v", logs.All())
|
||||
}
|
||||
}
|
||||
|
||||
// TestExchangeTraceQueryWrapsDatabaseErrors 验证查询故障不会降级为空关系。
|
||||
func TestExchangeTraceQueryWrapsDatabaseErrors(t *testing.T) {
|
||||
tx := testutil.NewPostgresTransaction(t)
|
||||
callbackName := "ur86:force_trace_query_error"
|
||||
if err := tx.Callback().Query().Before("gorm:query").Register(callbackName, func(db *gorm.DB) {
|
||||
db.AddError(fmt.Errorf("UR86 模拟数据库故障"))
|
||||
}); err != nil {
|
||||
t.Fatalf("注册数据库故障回调失败：%v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = tx.Callback().Query().Remove(callbackName) })
|
||||
|
||||
_, err := NewExchangeTraceQuery(tx, zap.NewNop()).Resolve(context.Background(), constants.ExchangeAssetTypeIotCard, 1)
|
||||
appErr, ok := err.(*errors.AppError)
|
||||
if !ok || appErr.Code != errors.CodeDatabaseError {
|
||||
t.Fatalf("数据库故障应转换为统一错误，实际：%v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestExchangeTraceQueryProjectsBidirectionalChain 验证 A→B→C 中间资产同时返回前代和后代。
|
||||
func TestExchangeTraceQueryProjectsBidirectionalChain(t *testing.T) {
|
||||
tx := testutil.NewPostgresTransaction(t)
|
||||
query := NewExchangeTraceQuery(tx, zap.NewNop())
|
||||
a := createTraceCard(t, tx, 51, nil)
|
||||
b := createTraceCard(t, tx, 52, nil)
|
||||
c := createTraceCard(t, tx, 53, nil)
|
||||
createTraceOrder(t, tx, "UR86-CHAIN-AB", constants.ExchangeAssetTypeIotCard, a.ID, "快照A", constants.ExchangeAssetTypeIotCard, b.ID, "快照B", constants.ExchangeStatusCompleted, time.Now().Add(-time.Minute))
|
||||
createTraceOrder(t, tx, "UR86-CHAIN-BC", constants.ExchangeAssetTypeIotCard, b.ID, "快照B", constants.ExchangeAssetTypeIotCard, c.ID, "快照C", constants.ExchangeStatusCompleted, time.Now())
|
||||
|
||||
trace, err := query.Resolve(context.Background(), constants.ExchangeAssetTypeIotCard, b.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("查询双向换货链失败：%v", err)
|
||||
}
|
||||
assertTraceAsset(t, trace.PreviousAsset, constants.ExchangeAssetTypeIotCard, a.ID, "快照A", "UR86-CHAIN-AB", true)
|
||||
assertTraceAsset(t, trace.NextAsset, constants.ExchangeAssetTypeIotCard, c.ID, "快照C", "UR86-CHAIN-BC", true)
|
||||
|
||||
trace, err = query.Resolve(context.Background(), constants.ExchangeAssetTypeIotCard, a.ID)
|
||||
if err != nil || trace.PreviousAsset != nil {
|
||||
t.Fatalf("链首不应有前代：trace=%+v err=%v", trace, err)
|
||||
}
|
||||
assertTraceAsset(t, trace.NextAsset, constants.ExchangeAssetTypeIotCard, b.ID, "快照B", "UR86-CHAIN-AB", true)
|
||||
}
|
||||
|
||||
// TestExchangeTraceQueryHidesInvisibleNextAsset 验证后代不可见时保留快照但隐藏 ID。
|
||||
func TestExchangeTraceQueryHidesInvisibleNextAsset(t *testing.T) {
|
||||
tx := testutil.NewPostgresTransaction(t)
|
||||
visibleShop, hiddenShop := uint(86201), uint(86202)
|
||||
oldDevice := createTraceDevice(t, tx, 61, &visibleShop)
|
||||
newDevice := createTraceDevice(t, tx, 62, &hiddenShop)
|
||||
createTraceOrder(t, tx, "UR86-NEXT-HIDDEN", constants.ExchangeAssetTypeDevice, oldDevice.ID, "旧设备快照", constants.ExchangeAssetTypeDevice, newDevice.ID, "不可见新设备快照", constants.ExchangeStatusCompleted, time.Now())
|
||||
ctx := context.WithValue(context.Background(), constants.ContextKeySubordinateShopIDs, []uint{visibleShop})
|
||||
|
||||
trace, err := NewExchangeTraceQuery(tx, zap.NewNop()).Resolve(ctx, constants.ExchangeAssetTypeDevice, oldDevice.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("查询不可见后代失败：%v", err)
|
||||
}
|
||||
assertTraceAsset(t, trace.NextAsset, constants.ExchangeAssetTypeDevice, 0, "不可见新设备快照", "UR86-NEXT-HIDDEN", false)
|
||||
}
|
||||
|
||||
// TestExchangeTraceQueryKeepsNextSnapshotWhenAssetIDMissing 验证历史完成单缺少新资产 ID 时仍保留后代文本。
|
||||
func TestExchangeTraceQueryKeepsNextSnapshotWhenAssetIDMissing(t *testing.T) {
|
||||
tx := testutil.NewPostgresTransaction(t)
|
||||
core, logs := observer.New(zap.WarnLevel)
|
||||
oldCard := createTraceCard(t, tx, 63, nil)
|
||||
completedAt := time.Now()
|
||||
order := &model.ExchangeOrder{
|
||||
ExchangeNo: "UR86-NEXT-MISSING-ID", FlowType: constants.ExchangeFlowTypeDirect,
|
||||
OldAssetType: constants.ExchangeAssetTypeIotCard, OldAssetID: oldCard.ID, OldAssetIdentifier: "旧快照",
|
||||
NewAssetType: constants.ExchangeAssetTypeIotCard, NewAssetIdentifier: "缺失 ID 的后代快照",
|
||||
ExchangeReason: "UR86 历史异常测试", Status: constants.ExchangeStatusCompleted, CompletedAt: &completedAt,
|
||||
}
|
||||
if err := tx.Create(order).Error; err != nil {
|
||||
t.Fatalf("创建缺少新资产 ID 的历史换货单失败：%v", err)
|
||||
}
|
||||
|
||||
trace, err := NewExchangeTraceQuery(tx, zap.New(core)).Resolve(context.Background(), constants.ExchangeAssetTypeIotCard, oldCard.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("查询缺少新资产 ID 的后代失败：%v", err)
|
||||
}
|
||||
assertTraceAsset(t, trace.NextAsset, constants.ExchangeAssetTypeIotCard, 0, "缺失 ID 的后代快照", "UR86-NEXT-MISSING-ID", false)
|
||||
if logs.FilterMessage("已完成换货记录缺少新资产 ID").Len() != 1 {
|
||||
t.Fatalf("应记录历史换货单缺少新资产 ID 的异常：%v", logs.All())
|
||||
}
|
||||
}
|
||||
|
||||
// TestExchangeTraceQueryProvidesBatchRelationshipQueries 验证前代和后代关系支持批量查询。
|
||||
func TestExchangeTraceQueryProvidesBatchRelationshipQueries(t *testing.T) {
|
||||
tx := testutil.NewPostgresTransaction(t)
|
||||
a := createTraceCard(t, tx, 64, nil)
|
||||
b := createTraceCard(t, tx, 65, nil)
|
||||
c := createTraceDevice(t, tx, 66, nil)
|
||||
d := createTraceDevice(t, tx, 67, nil)
|
||||
createTraceOrder(t, tx, "UR86-BATCH-CARD", constants.ExchangeAssetTypeIotCard, a.ID, "快照A", constants.ExchangeAssetTypeIotCard, b.ID, "快照B", constants.ExchangeStatusCompleted, time.Now())
|
||||
createTraceOrder(t, tx, "UR86-BATCH-DEVICE", constants.ExchangeAssetTypeDevice, c.ID, "快照C", constants.ExchangeAssetTypeDevice, d.ID, "快照D", constants.ExchangeStatusCompleted, time.Now())
|
||||
query := NewExchangeTraceQuery(tx, zap.NewNop())
|
||||
|
||||
previousRefs := []ExchangeTraceAssetRef{{AssetType: constants.AssetResolveTypeCard, AssetID: b.ID}, {AssetType: constants.ExchangeAssetTypeDevice, AssetID: d.ID}}
|
||||
previous, err := query.FindPreviousCompleted(context.Background(), previousRefs)
|
||||
if err != nil || len(previous[ExchangeTraceAssetRef{AssetType: constants.ExchangeAssetTypeIotCard, AssetID: b.ID}]) != 1 || len(previous[previousRefs[1]]) != 1 {
|
||||
t.Fatalf("批量前代查询错误：result=%+v err=%v", previous, err)
|
||||
}
|
||||
nextRefs := []ExchangeTraceAssetRef{{AssetType: constants.ExchangeAssetTypeIotCard, AssetID: a.ID}, {AssetType: constants.ExchangeAssetTypeDevice, AssetID: c.ID}}
|
||||
next, err := query.FindNextCompleted(context.Background(), nextRefs)
|
||||
if err != nil || len(next[nextRefs[0]]) != 1 || len(next[nextRefs[1]]) != 1 {
|
||||
t.Fatalf("批量后代查询错误：result=%+v err=%v", next, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestExchangeTraceQueryIgnoresIncompleteDeletedAndSelectsLatestNextRelation 验证后代过滤、确定性选择和异常日志。
|
||||
func TestExchangeTraceQueryIgnoresIncompleteDeletedAndSelectsLatestNextRelation(t *testing.T) {
|
||||
tx := testutil.NewPostgresTransaction(t)
|
||||
core, logs := observer.New(zap.WarnLevel)
|
||||
query := NewExchangeTraceQuery(tx, zap.New(core))
|
||||
oldDevice := createTraceDevice(t, tx, 71, nil)
|
||||
statuses := []int{constants.ExchangeStatusPendingInfo, constants.ExchangeStatusPendingShip, constants.ExchangeStatusShipped, constants.ExchangeStatusCancelled}
|
||||
for index, status := range statuses {
|
||||
candidate := createTraceDevice(t, tx, 72+index, nil)
|
||||
createTraceOrder(t, tx, fmt.Sprintf("UR86-NEXT-STATUS-%d", status), constants.ExchangeAssetTypeDevice, oldDevice.ID, "旧快照", constants.ExchangeAssetTypeDevice, candidate.ID, "候选快照", status, time.Now())
|
||||
}
|
||||
deletedCandidate := createTraceDevice(t, tx, 80, nil)
|
||||
deletedOrder := createTraceOrder(t, tx, "UR86-NEXT-DELETED", constants.ExchangeAssetTypeDevice, oldDevice.ID, "旧快照", constants.ExchangeAssetTypeDevice, deletedCandidate.ID, "软删除快照", constants.ExchangeStatusCompleted, time.Now())
|
||||
if err := tx.Delete(deletedOrder).Error; err != nil {
|
||||
t.Fatalf("软删除后代换货单失败：%v", err)
|
||||
}
|
||||
|
||||
trace, err := query.Resolve(context.Background(), constants.ExchangeAssetTypeDevice, oldDevice.ID)
|
||||
if err != nil || trace.NextAsset != nil {
|
||||
t.Fatalf("非完成和软删除记录不应形成后代：trace=%+v err=%v", trace, err)
|
||||
}
|
||||
|
||||
newOne := createTraceDevice(t, tx, 81, nil)
|
||||
newTwo := createTraceDevice(t, tx, 82, nil)
|
||||
completedAt := time.Now().Truncate(time.Second)
|
||||
createTraceOrder(t, tx, "UR86-NEXT-OLDER", constants.ExchangeAssetTypeDevice, oldDevice.ID, "旧快照", constants.ExchangeAssetTypeDevice, newOne.ID, "较旧后代快照", constants.ExchangeStatusCompleted, completedAt)
|
||||
createTraceOrder(t, tx, "UR86-NEXT-LATEST", constants.ExchangeAssetTypeDevice, oldDevice.ID, "旧快照", constants.ExchangeAssetTypeDevice, newTwo.ID, "最新后代快照", constants.ExchangeStatusCompleted, completedAt)
|
||||
trace, err = query.Resolve(context.Background(), constants.ExchangeAssetTypeDevice, oldDevice.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("查询重复后代记录失败：%v", err)
|
||||
}
|
||||
assertTraceAsset(t, trace.NextAsset, constants.ExchangeAssetTypeDevice, newTwo.ID, "最新后代快照", "UR86-NEXT-LATEST", true)
|
||||
if logs.FilterMessage("检测到同方向多条已完成换货记录").Len() != 1 {
|
||||
t.Fatalf("应记录后代重复换货异常日志：%v", logs.All())
|
||||
}
|
||||
}
|
||||
|
||||
// TestExchangeTraceQueryUsesFixedQueriesAndMeetsPerformanceTargets 验证 SQL 次数固定且读取性能满足项目目标。
|
||||
func TestExchangeTraceQueryUsesFixedQueriesAndMeetsPerformanceTargets(t *testing.T) {
|
||||
tx := testutil.NewPostgresTransaction(t)
|
||||
a := createTraceCard(t, tx, 91, nil)
|
||||
b := createTraceCard(t, tx, 92, nil)
|
||||
c := createTraceCard(t, tx, 93, nil)
|
||||
createTraceOrder(t, tx, "UR86-PERF-AB", constants.ExchangeAssetTypeIotCard, a.ID, "快照A", constants.ExchangeAssetTypeIotCard, b.ID, "快照B", constants.ExchangeStatusCompleted, time.Now().Add(-time.Minute))
|
||||
createTraceOrder(t, tx, "UR86-PERF-BC", constants.ExchangeAssetTypeIotCard, b.ID, "快照B", constants.ExchangeAssetTypeIotCard, c.ID, "快照C", constants.ExchangeStatusCompleted, time.Now())
|
||||
counter := &traceQueryCounter{Interface: tx.Logger}
|
||||
query := NewExchangeTraceQuery(tx.Session(&gorm.Session{Logger: counter}), zap.NewNop())
|
||||
|
||||
durations := make([]time.Duration, 30)
|
||||
for index := range durations {
|
||||
startedAt := time.Now()
|
||||
trace, err := query.Resolve(context.Background(), constants.ExchangeAssetTypeIotCard, b.ID)
|
||||
durations[index] = time.Since(startedAt)
|
||||
if err != nil || trace.PreviousAsset == nil || trace.NextAsset == nil {
|
||||
t.Fatalf("第 %d 次性能查询失败：trace=%+v err=%v", index+1, trace, err)
|
||||
}
|
||||
}
|
||||
if counter.count.Load() != 90 {
|
||||
t.Fatalf("卡链每次应固定执行 3 条 SQL，实际总数：%d", counter.count.Load())
|
||||
}
|
||||
sort.Slice(durations, func(left, right int) bool { return durations[left] < durations[right] })
|
||||
p95, p99 := durations[28], durations[29]
|
||||
if p95 >= 200*time.Millisecond || p99 >= 500*time.Millisecond {
|
||||
t.Fatalf("换货链 Query 超过 API 性能目标：p95=%s p99=%s", p95, p99)
|
||||
}
|
||||
}
|
||||
|
||||
// TestExchangeTraceIndexDefinitionAndPlans 验证双向部分索引定义及代表性查询计划。
|
||||
func TestExchangeTraceIndexDefinitionAndPlans(t *testing.T) {
|
||||
tx := testutil.NewPostgresTransaction(t)
|
||||
indexNames := []string{"idx_exchange_trace_new_asset", "idx_exchange_trace_old_asset"}
|
||||
for _, indexName := range indexNames {
|
||||
var count int64
|
||||
if err := tx.Raw("SELECT COUNT(*) FROM pg_class WHERE relname = ?", indexName).Scan(&count).Error; err != nil {
|
||||
t.Fatalf("查询换货链索引数量失败：%v", err)
|
||||
}
|
||||
if os.Getenv("UR86_EXPECT_INDEX_ABSENT") == "1" {
|
||||
if count != 0 {
|
||||
t.Fatalf("回滚后索引 %s 仍然存在", indexName)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if count != 1 {
|
||||
t.Fatalf("换货链索引 %s 数量异常：%d", indexName, count)
|
||||
}
|
||||
|
||||
var index struct {
|
||||
AccessMethod string `gorm:"column:access_method"`
|
||||
IsUnique bool `gorm:"column:is_unique"`
|
||||
Definition string `gorm:"column:definition"`
|
||||
Predicate string `gorm:"column:predicate"`
|
||||
}
|
||||
err := tx.Raw(`
|
||||
SELECT am.amname AS access_method,
|
||||
ix.indisunique AS is_unique,
|
||||
pg_get_indexdef(ix.indexrelid) AS definition,
|
||||
pg_get_expr(ix.indpred, ix.indrelid) AS predicate
|
||||
FROM pg_index ix
|
||||
JOIN pg_class i ON i.oid = ix.indexrelid
|
||||
JOIN pg_am am ON am.oid = i.relam
|
||||
WHERE i.relname = ?
|
||||
`, indexName).Scan(&index).Error
|
||||
if err != nil {
|
||||
t.Fatalf("查询换货链索引定义失败：%v", err)
|
||||
}
|
||||
if index.AccessMethod != "btree" || index.IsUnique || !strings.Contains(index.Definition, "completed_at DESC NULLS LAST") || !strings.Contains(index.Definition, "id DESC") || !strings.Contains(index.Predicate, "status = 4") || !strings.Contains(index.Predicate, "deleted_at IS NULL") {
|
||||
t.Fatalf("换货链索引定义不符合契约：%+v", index)
|
||||
}
|
||||
}
|
||||
if os.Getenv("UR86_EXPECT_INDEX_ABSENT") == "1" {
|
||||
return
|
||||
}
|
||||
|
||||
for direction, sql := range map[string]string{
|
||||
"previous": "SELECT * FROM tb_exchange_order WHERE new_asset_type = 'iot_card' AND new_asset_id = 1 AND status = 4 AND deleted_at IS NULL ORDER BY completed_at DESC NULLS LAST, id DESC LIMIT 1",
|
||||
"next": "SELECT * FROM tb_exchange_order WHERE old_asset_type = 'iot_card' AND old_asset_id = 1 AND status = 4 AND deleted_at IS NULL ORDER BY completed_at DESC NULLS LAST, id DESC LIMIT 1",
|
||||
} {
|
||||
var planLines []string
|
||||
if err := tx.Exec("SET LOCAL enable_seqscan = off").Error; err != nil {
|
||||
t.Fatalf("配置查询计划测试失败：%v", err)
|
||||
}
|
||||
if err := tx.Raw("EXPLAIN " + sql).Scan(&planLines).Error; err != nil {
|
||||
t.Fatalf("记录 %s 查询计划失败：%v", direction, err)
|
||||
}
|
||||
plan := strings.Join(planLines, "\n")
|
||||
expectedIndex := "idx_exchange_trace_old_asset"
|
||||
if direction == "previous" {
|
||||
expectedIndex = "idx_exchange_trace_new_asset"
|
||||
}
|
||||
if !strings.Contains(plan, expectedIndex) {
|
||||
t.Fatalf("%s 查询计划未使用预期索引 %s：%s", direction, expectedIndex, plan)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func createTraceCard(t *testing.T, tx *gorm.DB, suffix int, shopID *uint) *model.IotCard {
|
||||
t.Helper()
|
||||
iccid := fmt.Sprintf("8986222222222222%04d", suffix)
|
||||
card := &model.IotCard{ICCID: iccid, ICCID19: iccid[:19], VirtualNo: fmt.Sprintf("UR86-CARD-%04d", suffix), ShopID: shopID, AssetStatus: constants.AssetStatusInStock}
|
||||
if err := tx.Create(card).Error; err != nil {
|
||||
t.Fatalf("创建换货链测试卡失败：%v", err)
|
||||
}
|
||||
return card
|
||||
}
|
||||
|
||||
func createTraceDevice(t *testing.T, tx *gorm.DB, suffix int, shopID *uint) *model.Device {
|
||||
t.Helper()
|
||||
device := &model.Device{VirtualNo: fmt.Sprintf("UR86-DEVICE-%04d", suffix), IMEI: fmt.Sprintf("86222222222%04d", suffix), SN: fmt.Sprintf("UR86-SN-%04d", suffix), ShopID: shopID, AssetStatus: constants.AssetStatusInStock}
|
||||
if err := tx.Create(device).Error; err != nil {
|
||||
t.Fatalf("创建换货链测试设备失败：%v", err)
|
||||
}
|
||||
return device
|
||||
}
|
||||
|
||||
func createTraceOrder(t *testing.T, tx *gorm.DB, exchangeNo, oldType string, oldID uint, oldIdentifier, newType string, newID uint, newIdentifier string, status int, completedAt time.Time) *model.ExchangeOrder {
|
||||
t.Helper()
|
||||
order := &model.ExchangeOrder{
|
||||
ExchangeNo: exchangeNo, FlowType: constants.ExchangeFlowTypeDirect,
|
||||
OldAssetType: oldType, OldAssetID: oldID, OldAssetIdentifier: oldIdentifier,
|
||||
NewAssetType: newType, NewAssetID: &newID, NewAssetIdentifier: newIdentifier,
|
||||
ExchangeReason: "UR86 换货链测试", Status: status,
|
||||
}
|
||||
if status == constants.ExchangeStatusCompleted {
|
||||
order.CompletedAt = &completedAt
|
||||
}
|
||||
if err := tx.Create(order).Error; err != nil {
|
||||
t.Fatalf("创建换货链测试单失败：%v", err)
|
||||
}
|
||||
return order
|
||||
}
|
||||
|
||||
func assertTraceAsset(t *testing.T, actual *dto.AssetExchangeTraceItem, assetType string, assetID uint, identifier, exchangeNo string, canView bool) {
|
||||
t.Helper()
|
||||
if actual == nil {
|
||||
t.Fatal("换货关联项不应为空")
|
||||
}
|
||||
if actual.AssetType != assetType || actual.Identifier != identifier || actual.ExchangeNo != exchangeNo || actual.CanView != canView {
|
||||
t.Fatalf("换货关联项错误：%+v", actual)
|
||||
}
|
||||
if canView {
|
||||
if actual.AssetID == nil || *actual.AssetID != assetID {
|
||||
t.Fatalf("可见关联资产应返回真实 ID：%+v", actual)
|
||||
}
|
||||
} else if actual.AssetID != nil {
|
||||
t.Fatalf("不可见关联资产不应返回 ID：%+v", actual)
|
||||
}
|
||||
}
|
||||
|
||||
type traceQueryCounter struct {
|
||||
logger.Interface
|
||||
count atomic.Int64
|
||||
}
|
||||
|
||||
func (l *traceQueryCounter) Trace(ctx context.Context, begin time.Time, fc func() (string, int64), err error) {
|
||||
l.count.Add(1)
|
||||
l.Interface.Trace(ctx, begin, fc, err)
|
||||
}
|
||||
Reference in New Issue
Block a user