直接换货流程

2026-06-03 16:55:32 +08:00
parent 46ede81aef
commit 5f57429fb0
13 changed files with 718 additions and 167 deletions
--- a/internal/middleware/personal_auth.go
+++ b/internal/middleware/personal_auth.go
@@ -7,6 +7,7 @@ import (
 	"github.com/break/junhong_cmp_fiber/pkg/auth"
 	"github.com/break/junhong_cmp_fiber/pkg/constants"
 	"github.com/break/junhong_cmp_fiber/pkg/errors"
+	pkgmiddleware "github.com/break/junhong_cmp_fiber/pkg/middleware"
 	"github.com/gofiber/fiber/v2"
 	"github.com/redis/go-redis/v9"
 	"go.uber.org/zap"
@@ -97,6 +98,9 @@ func (m *PersonalAuthMiddleware) Authenticate() fiber.Handler {
 		c.Locals("customer_asset_type", claims.AssetType)
 		c.Locals("customer_asset_id", claims.AssetID)
 		c.Locals("skip_owner_filter", true)
+		c.SetUserContext(pkgmiddleware.SetUserContext(c.UserContext(), &pkgmiddleware.UserContextInfo{
+			CustomerID: claims.CustomerID,
+		}))

 		m.logger.Debug("个人客户认证成功",
 			zap.Uint("customer_id", claims.CustomerID),
--- a/internal/model/dto/asset_wallet_dto.go
+++ b/internal/model/dto/asset_wallet_dto.go
@@ -22,7 +22,7 @@ type AssetWalletTransactionListRequest struct {
 	Identifier      string     `path:"identifier" description:"资产标识符（ICCID 或 VirtualNo）" required:"true"`
 	Page            int        `json:"page" query:"page" description:"页码，默认1"`
 	PageSize        int        `json:"page_size" query:"page_size" validate:"omitempty,min=1,max=100" description:"每页数量，默认20，最大100"`
-	TransactionType *string    `json:"transaction_type" query:"transaction_type" validate:"omitempty,oneof=recharge deduct refund" description:"交易类型过滤：recharge/deduct/refund"`
+	TransactionType *string    `json:"transaction_type" query:"transaction_type" validate:"omitempty,oneof=recharge deduct refund exchange" description:"交易类型过滤：recharge/deduct/refund/exchange"`
 	StartTime       *time.Time `json:"start_time" query:"start_time" description:"开始时间（RFC3339）"`
 	EndTime         *time.Time `json:"end_time" query:"end_time" description:"结束时间（RFC3339）"`
 }
@@ -30,12 +30,12 @@ type AssetWalletTransactionListRequest struct {
 // AssetWalletTransactionItem 单条流水记录
 type AssetWalletTransactionItem struct {
 	ID                  uint      `json:"id" description:"流水记录ID"`
-	TransactionType     string    `json:"transaction_type" description:"交易类型：recharge/deduct/refund"`
-	TransactionTypeText string    `json:"transaction_type_text" description:"交易类型文本：充值/扣款/退款"`
+	TransactionType     string    `json:"transaction_type" description:"交易类型：recharge/deduct/refund/exchange"`
+	TransactionTypeText string    `json:"transaction_type_text" description:"交易类型文本：充值/扣款/退款/换货迁移"`
 	Amount              int64     `json:"amount" description:"变动金额（分），充值为正数，扣款/退款为负数"`
 	BalanceBefore       int64     `json:"balance_before" description:"变动前余额（分）"`
 	BalanceAfter        int64     `json:"balance_after" description:"变动后余额（分）"`
-	ReferenceType       *string   `json:"reference_type,omitempty" description:"关联业务类型：recharge 或 order（可空）"`
+	ReferenceType       *string   `json:"reference_type,omitempty" description:"关联业务类型：recharge、order 或 exchange（可空）"`
 	ReferenceNo         *string   `json:"reference_no,omitempty" description:"关联业务编号：充值单号（CRCH…）或订单号（ORD…）（可空）"`
 	Remark              *string   `json:"remark,omitempty" description:"备注（可空）"`
 	CreatedAt           time.Time `json:"created_at" description:"流水创建时间（RFC3339）"`
--- a/internal/model/dto/exchange_dto.go
+++ b/internal/model/dto/exchange_dto.go
@@ -5,6 +5,9 @@ import "time"
 type CreateExchangeRequest struct {
 	OldAssetType   string  `json:"old_asset_type" validate:"required,oneof=iot_card device" required:"true" description:"旧资产类型 (iot_card:物联网卡, device:设备)"`
 	OldIdentifier  string  `json:"old_identifier" validate:"required,min=1,max=100" required:"true" minLength:"1" maxLength:"100" description:"旧资产标识符(ICCID/虚拟号/IMEI/SN)"`
+	FlowType       string  `json:"flow_type" validate:"omitempty,oneof=shipping direct" enum:"shipping,direct" description:"换货流程类型 (shipping:物流换货, direct:直接换货)"`
+	NewIdentifier  string  `json:"new_identifier" validate:"omitempty,min=1,max=100" minLength:"1" maxLength:"100" description:"新资产标识符，direct 流程必填(ICCID/虚拟号/IMEI/SN)"`
+	MigrateData    *bool   `json:"migrate_data" description:"是否执行全量迁移，direct 流程未传按 false 处理"`
 	ExchangeReason string  `json:"exchange_reason" validate:"required,min=1,max=100" required:"true" minLength:"1" maxLength:"100" description:"换货原因"`
 	Remark         *string `json:"remark" validate:"omitempty,max=500" maxLength:"500" description:"备注"`
 }
@@ -13,6 +16,7 @@ type ExchangeListRequest struct {
 	Page           int        `json:"page" query:"page" validate:"omitempty,min=1" minimum:"1" description:"页码"`
 	PageSize       int        `json:"page_size" query:"page_size" validate:"omitempty,min=1,max=100" minimum:"1" maximum:"100" description:"每页数量"`
 	Status         *int       `json:"status" query:"status" validate:"omitempty,min=1,max=5" minimum:"1" maximum:"5" description:"换货状态 (1:待填写信息, 2:待发货, 3:已发货待确认, 4:已完成, 5:已取消)"`
+	FlowType       string     `json:"flow_type" query:"flow_type" validate:"omitempty,oneof=shipping direct" enum:"shipping,direct" description:"换货流程类型 (shipping:物流换货, direct:直接换货)"`
 	Identifier     string     `json:"identifier" query:"identifier" validate:"omitempty,max=100" maxLength:"100" description:"资产标识符搜索(旧资产/新资产标识符模糊匹配)"`
 	CreatedAtStart *time.Time `json:"created_at_start" query:"created_at_start" description:"创建时间起始"`
 	CreatedAtEnd   *time.Time `json:"created_at_end" query:"created_at_end" description:"创建时间结束"`
@@ -61,6 +65,8 @@ type ClientShippingInfoParams struct {
 type ExchangeOrderResponse struct {
 	ID                 uint       `json:"id" description:"换货单ID"`
 	ExchangeNo         string     `json:"exchange_no" description:"换货单号"`
+	FlowType           string     `json:"flow_type" description:"换货流程类型 (shipping:物流换货, direct:直接换货)"`
+	FlowTypeName       string     `json:"flow_type_name" description:"换货流程类型名称"`
 	OldAssetType       string     `json:"old_asset_type" description:"旧资产类型 (iot_card:物联网卡, device:设备)"`
 	OldAssetID         uint       `json:"old_asset_id" description:"旧资产ID"`
 	OldAssetIdentifier string     `json:"old_asset_identifier" description:"旧资产标识符"`
@@ -75,9 +81,12 @@ type ExchangeOrderResponse struct {
 	MigrateData        bool       `json:"migrate_data" description:"是否执行全量迁移"`
 	MigrationCompleted bool       `json:"migration_completed" description:"迁移是否已完成"`
 	MigrationBalance   int64      `json:"migration_balance" description:"迁移转移金额(分)"`
+	ShippedAt          *time.Time `json:"shipped_at,omitempty" description:"发货时间"`
+	CompletedAt        *time.Time `json:"completed_at,omitempty" description:"换货完成时间"`
 	ExchangeReason     string     `json:"exchange_reason" description:"换货原因"`
 	Remark             *string    `json:"remark,omitempty" description:"备注"`
 	Status             int        `json:"status" description:"换货状态 (1:待填写信息, 2:待发货, 3:已发货待确认, 4:已完成, 5:已取消)"`
+	StatusName         string     `json:"status_name" description:"换货状态名称（中文）"`
 	StatusText         string     `json:"status_text" description:"换货状态文本"`
 	ShopID             *uint      `json:"shop_id,omitempty" description:"所属店铺ID"`
 	CreatedAt          time.Time  `json:"created_at" description:"创建时间"`
@@ -97,7 +106,9 @@ type ExchangeListResponse struct {
 type ClientExchangePendingResponse struct {
 	ID             uint      `json:"id" description:"换货单ID"`
 	ExchangeNo     string    `json:"exchange_no" description:"换货单号"`
+	FlowType       string    `json:"flow_type" description:"换货流程类型 (shipping:物流换货, direct:直接换货)"`
 	Status         int       `json:"status" description:"换货状态 (1:待填写信息, 2:待发货, 3:已发货待确认, 4:已完成, 5:已取消)"`
+	StatusName     string    `json:"status_name" description:"换货状态名称（中文）"`
 	StatusText     string    `json:"status_text" description:"换货状态文本"`
 	ExchangeReason string    `json:"exchange_reason" description:"换货原因"`
 	CreatedAt      time.Time `json:"created_at" description:"创建时间"`
--- a/internal/model/exchange_order.go
+++ b/internal/model/exchange_order.go
@@ -9,8 +9,8 @@ import (
 )

 // ExchangeOrder 换货单模型
-// 承载客户端换货的完整生命周期：后台发起 → 客户端填写收货信息 → 后台发货 → 确认完成（含可选全量迁移） → 旧资产可转新
-// 状态机：1-待填写信息 → 2-待发货 → 3-已发货待确认 → 4-已完成，1/2 时可取消 → 5-已取消
+// 承载物流换货与直接换货的完整生命周期：后台发起 → 可选客户填写收货信息 → 可选后台发货 → 确认完成（含可选全量迁移） → 旧资产可转新
+// 状态机：shipping 为 1-待填写信息 → 2-待发货 → 3-已发货待确认 → 4-已完成，1/2 时可取消 → 5-已取消；direct 创建即完成为 4
 type ExchangeOrder struct {
 	gorm.Model
 	BaseModel `gorm:"embedded"`
@@ -18,6 +18,9 @@ type ExchangeOrder struct {
 	// 单号
 	ExchangeNo string `gorm:"column:exchange_no;type:varchar(50);not null;uniqueIndex:idx_exchange_order_no,where:deleted_at IS NULL;comment:换货单号(EXC+日期+随机数)" json:"exchange_no"`

+	// 流程类型
+	FlowType string `gorm:"column:flow_type;type:varchar(20);not null;default:'shipping';index:idx_exchange_order_flow_status;comment:换货流程类型 shipping-物流换货 direct-直接换货" json:"flow_type"`
+
 	// 旧资产快照
 	OldAssetType       string `gorm:"column:old_asset_type;type:varchar(20);not null;comment:旧资产类型(iot_card/device)" json:"old_asset_type"`
 	OldAssetID         uint   `gorm:"column:old_asset_id;not null;index:idx_exchange_order_old_asset;comment:旧资产ID" json:"old_asset_id"`
@@ -42,6 +45,10 @@ type ExchangeOrder struct {
 	MigrationCompleted bool  `gorm:"column:migration_completed;type:boolean;default:false;comment:迁移是否已完成" json:"migration_completed"`
 	MigrationBalance   int64 `gorm:"column:migration_balance;type:bigint;default:0;comment:迁移转移金额(分)" json:"migration_balance"`

+	// 业务时间
+	ShippedAt   *time.Time `gorm:"column:shipped_at;comment:发货时间，仅物流换货发货成功后写入" json:"shipped_at,omitempty"`
+	CompletedAt *time.Time `gorm:"column:completed_at;comment:换货完成时间，物流换货和直接换货完成后写入" json:"completed_at,omitempty"`
+
 	// 业务信息
 	ExchangeReason string  `gorm:"column:exchange_reason;type:varchar(100);not null;comment:换货原因" json:"exchange_reason"`
 	Remark         *string `gorm:"column:remark;type:text;comment:备注" json:"remark,omitempty"`
--- a/internal/service/asset/service.go
+++ b/internal/service/asset/service.go
@@ -1016,13 +1016,17 @@ func (s *Service) tracePreviousGenerations(ctx context.Context, assetType string
 		}

 		oldGeneration := s.resolveAssetGeneration(ctx, oldAssetType, exchOrder.OldAssetID)
+		exchangedAt := exchOrder.UpdatedAt
+		if exchOrder.CompletedAt != nil && !exchOrder.CompletedAt.IsZero() {
+			exchangedAt = *exchOrder.CompletedAt
+		}

 		prevGens = append(prevGens, &dto.PreviousGenerationOrders{
 			Generation:  oldGeneration,
 			Identifier:  exchOrder.OldAssetIdentifier,
 			AssetType:   oldAssetType,
 			ExchangeNo:  exchOrder.ExchangeNo,
-			ExchangedAt: exchOrder.UpdatedAt,
+			ExchangedAt: exchangedAt,
 			Total:       oldTotal,
 			Items:       orderItems,
 		})
--- a/internal/service/asset_wallet/service.go
+++ b/internal/service/asset_wallet/service.go
@@ -5,6 +5,7 @@ import (

 	"github.com/break/junhong_cmp_fiber/internal/model/dto"
 	"github.com/break/junhong_cmp_fiber/internal/store/postgres"
+	"github.com/break/junhong_cmp_fiber/pkg/constants"
 	"github.com/break/junhong_cmp_fiber/pkg/errors"
 	"gorm.io/gorm"
 )
@@ -145,12 +146,14 @@ func walletStatusText(status int) string {
 // transactionTypeText 翻译交易类型文本
 func transactionTypeText(transactionType string) string {
 	switch transactionType {
-	case "recharge":
+	case constants.AssetTransactionTypeRecharge:
 		return "充值"
-	case "deduct":
+	case constants.AssetTransactionTypeDeduct:
 		return "扣款"
-	case "refund":
+	case constants.AssetTransactionTypeRefund:
 		return "退款"
+	case constants.AssetTransactionTypeExchange:
+		return "换货迁移"
 	default:
 		return transactionType
 	}
--- a/internal/service/exchange/migration.go
+++ b/internal/service/exchange/migration.go
@@ -12,67 +12,20 @@ import (
 	"gorm.io/gorm/clause"
 )

-func (s *Service) executeMigration(ctx context.Context, order *model.ExchangeOrder) (int64, error) {
-	var migrationBalance int64
-
-	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
-		if order.NewAssetID == nil || *order.NewAssetID == 0 {
-			return errors.New(errors.CodeInvalidParam, "新资产信息缺失")
-		}
-
-		oldAsset, err := s.resolveAssetByIdentifier(ctx, order.OldAssetType, order.OldAssetIdentifier)
-		if err != nil {
-			return err
-		}
-		newAsset, err := s.resolveAssetByIdentifier(ctx, order.OldAssetType, order.NewAssetIdentifier)
-		if err != nil {
-			return err
-		}
-
-		migrationBalance, err = s.transferWalletBalanceWithTx(ctx, tx, order, oldAsset, newAsset)
-		if err != nil {
-			return err
-		}
-
-		if err = s.migratePackageUsageWithTx(ctx, tx, oldAsset, newAsset); err != nil {
-			return err
-		}
-
-		if err = s.copyAccumulatedFieldsWithTx(tx, oldAsset, newAsset); err != nil {
-			return err
-		}
-
-		if err = s.copyResourceTagsWithTx(ctx, tx, oldAsset, newAsset); err != nil {
-			return err
-		}
-
-		if oldAsset.VirtualNo != "" && newAsset.VirtualNo != "" {
-			if err = tx.Model(&model.PersonalCustomerDevice{}).
-				Where("virtual_no = ?", oldAsset.VirtualNo).
-				Updates(map[string]any{"virtual_no": newAsset.VirtualNo, "updated_at": time.Now()}).Error; err != nil {
-				return errors.Wrap(errors.CodeDatabaseError, err, "更新客户绑定关系失败")
-			}
-		}
-
-		if err = s.updateOldAssetStatusWithTx(tx, oldAsset); err != nil {
-			return err
-		}
-
-		if err = tx.Model(&model.ExchangeOrder{}).Where("id = ?", order.ID).Updates(map[string]any{
-			"migration_completed": true,
-			"migration_balance":   migrationBalance,
-			"updater":             middleware.GetUserIDFromContext(ctx),
-			"updated_at":          time.Now(),
-		}).Error; err != nil {
-			return errors.Wrap(errors.CodeDatabaseError, err, "更新换货单迁移状态失败")
-		}
-
-		return nil
-	})
+func (s *Service) executeMigrationWithTx(ctx context.Context, tx *gorm.DB, order *model.ExchangeOrder, oldAsset, newAsset *resolvedExchangeAsset) (int64, error) {
+	migrationBalance, err := s.transferWalletBalanceWithTx(ctx, tx, order, oldAsset, newAsset)
 	if err != nil {
-		return 0, errors.Wrap(errors.CodeExchangeMigrationFailed, err, "执行全量迁移失败")
+		return 0, errors.Wrap(errors.CodeExchangeMigrationFailed, err, "执行钱包迁移失败")
+	}
+	if err = s.migratePackageUsageWithTx(ctx, tx, oldAsset, newAsset); err != nil {
+		return 0, errors.Wrap(errors.CodeExchangeMigrationFailed, err, "迁移套餐使用记录失败")
+	}
+	if err = s.copyAccumulatedFieldsWithTx(tx, oldAsset, newAsset); err != nil {
+		return 0, errors.Wrap(errors.CodeExchangeMigrationFailed, err, "复制累计充值字段失败")
+	}
+	if err = s.copyResourceTagsWithTx(ctx, tx, oldAsset, newAsset); err != nil {
+		return 0, errors.Wrap(errors.CodeExchangeMigrationFailed, err, "复制资产标签失败")
 	}
-
 	return migrationBalance, nil
 }

@@ -83,6 +36,9 @@ func (s *Service) transferWalletBalanceWithTx(ctx context.Context, tx *gorm.DB,
 			return 0, errors.Wrap(errors.CodeDatabaseError, err, "查询旧资产钱包失败")
 		}
 	}
+	if oldWallet.ID > 0 && oldWallet.FrozenBalance > 0 {
+		return 0, errors.New(errors.CodeExchangeMigrationFailed, "旧资产钱包存在冻结余额，请先处理冻结业务")
+	}

 	var newWallet model.AssetWallet
 	if err := tx.WithContext(ctx).Where("resource_type = ? AND resource_id = ?", newAsset.AssetType, newAsset.AssetID).First(&newWallet).Error; err != nil {
@@ -94,7 +50,7 @@ func (s *Service) transferWalletBalanceWithTx(ctx context.Context, tx *gorm.DB,
 		if newAsset.ShopID != nil {
 			shopTag = *newAsset.ShopID
 		}
-		newWallet = model.AssetWallet{ResourceType: newAsset.AssetType, ResourceID: newAsset.AssetID, Balance: 0, FrozenBalance: 0, Currency: "CNY", Status: 1, Version: 0, ShopIDTag: shopTag}
+		newWallet = model.AssetWallet{ResourceType: newAsset.AssetType, ResourceID: newAsset.AssetID, Balance: 0, FrozenBalance: 0, Currency: "CNY", Status: constants.AssetWalletStatusNormal, Version: 0, ShopIDTag: shopTag}
 		if err = tx.WithContext(ctx).Create(&newWallet).Error; err != nil {
 			return 0, errors.Wrap(errors.CodeDatabaseError, err, "创建新资产钱包失败")
 		}
@@ -114,19 +70,21 @@ func (s *Service) transferWalletBalanceWithTx(ctx context.Context, tx *gorm.DB,
 		return 0, errors.Wrap(errors.CodeDatabaseError, err, "增加新资产钱包余额失败")
 	}

-	refType := "exchange"
+	refType := constants.ReferenceTypeExchange
+	remark := "换货余额迁移"
 	if err := tx.WithContext(ctx).Create(&model.AssetWalletTransaction{
 		AssetWalletID:   newWallet.ID,
 		ResourceType:    newAsset.AssetType,
 		ResourceID:      newAsset.AssetID,
 		UserID:          middleware.GetUserIDFromContext(ctx),
-		TransactionType: "refund",
+		TransactionType: constants.AssetTransactionTypeExchange,
 		Amount:          migrationBalance,
 		BalanceBefore:   beforeBalance,
 		BalanceAfter:    beforeBalance + migrationBalance,
-		Status:          1,
+		Status:          constants.TransactionStatusSuccess,
 		ReferenceType:   &refType,
 		ReferenceNo:     &order.ExchangeNo,
+		Remark:          &remark,
 		Creator:         middleware.GetUserIDFromContext(ctx),
 		ShopIDTag:       newWallet.ShopIDTag,
 		EnterpriseIDTag: newWallet.EnterpriseIDTag,
@@ -224,16 +182,3 @@ func (s *Service) copyResourceTagsWithTx(ctx context.Context, tx *gorm.DB, oldAs
 	}
 	return nil
 }
-
-func (s *Service) updateOldAssetStatusWithTx(tx *gorm.DB, oldAsset *resolvedExchangeAsset) error {
-	if oldAsset.AssetType == constants.ExchangeAssetTypeIotCard {
-		if err := tx.Model(&model.IotCard{}).Where("id = ?", oldAsset.AssetID).Updates(map[string]any{"asset_status": 3, "updated_at": time.Now()}).Error; err != nil {
-			return errors.Wrap(errors.CodeDatabaseError, err, "更新旧卡状态失败")
-		}
-		return nil
-	}
-	if err := tx.Model(&model.Device{}).Where("id = ?", oldAsset.AssetID).Updates(map[string]any{"asset_status": 3, "updated_at": time.Now()}).Error; err != nil {
-		return errors.Wrap(errors.CodeDatabaseError, err, "更新旧设备状态失败")
-	}
-	return nil
-}
--- a/internal/service/exchange/service.go
+++ b/internal/service/exchange/service.go
@@ -2,6 +2,7 @@ package exchange

 import (
 	"context"
+	"strings"
 	"time"

 	"github.com/break/junhong_cmp_fiber/internal/model"
@@ -12,6 +13,7 @@ import (
 	"github.com/break/junhong_cmp_fiber/pkg/middleware"
 	"go.uber.org/zap"
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 )

 type Service struct {
@@ -57,10 +59,21 @@ func New(
 }

 func (s *Service) Create(ctx context.Context, req *dto.CreateExchangeRequest) (*dto.ExchangeOrderResponse, error) {
+	flowType := normalizeExchangeFlowType(req.FlowType)
+	if !isValidExchangeFlowType(flowType) {
+		return nil, errors.New(errors.CodeInvalidParam, "换货流程类型不合法")
+	}
+	if flowType == constants.ExchangeFlowTypeDirect && strings.TrimSpace(req.NewIdentifier) == "" {
+		return nil, errors.New(errors.CodeInvalidParam, "直接换货必须填写新资产标识")
+	}
+
 	asset, err := s.resolveAssetByIdentifier(ctx, req.OldAssetType, req.OldIdentifier)
 	if err != nil {
 		return nil, err
 	}
+	if asset.AssetStatus != constants.AssetStatusSold {
+		return nil, errors.New(errors.CodeExchangeStatusInvalid, "旧资产状态不允许换货")
+	}

 	if _, err = s.exchangeStore.FindActiveByOldAsset(ctx, asset.AssetType, asset.AssetID); err == nil {
 		return nil, errors.New(errors.CodeExchangeInProgress)
@@ -68,10 +81,14 @@ func (s *Service) Create(ctx context.Context, req *dto.CreateExchangeRequest) (*
 		return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询进行中换货单失败")
 	}

-	shopID := middleware.GetShopIDFromContext(ctx)
+	if flowType == constants.ExchangeFlowTypeDirect {
+		return s.createDirectExchange(ctx, req, asset)
+	}
+
 	creator := middleware.GetUserIDFromContext(ctx)
 	order := &model.ExchangeOrder{
 		ExchangeNo:         model.GenerateExchangeNo(),
+		FlowType:           constants.ExchangeFlowTypeShipping,
 		OldAssetType:       asset.AssetType,
 		OldAssetID:         asset.AssetID,
 		OldAssetIdentifier: asset.Identifier,
@@ -83,8 +100,8 @@ func (s *Service) Create(ctx context.Context, req *dto.CreateExchangeRequest) (*
 		MigrateData:        false,
 		BaseModel:          model.BaseModel{Creator: creator, Updater: creator},
 	}
-	if shopID > 0 {
-		order.ShopID = &shopID
+	if asset.ShopID != nil {
+		order.ShopID = asset.ShopID
 	}

 	if err = s.exchangeStore.Create(ctx, order); err != nil {
@@ -109,6 +126,9 @@ func (s *Service) List(ctx context.Context, req *dto.ExchangeListRequest) (*dto.
 	if req.Status != nil {
 		filters["status"] = *req.Status
 	}
+	if req.FlowType != "" {
+		filters["flow_type"] = normalizeExchangeFlowType(req.FlowType)
+	}
 	if req.Identifier != "" {
 		filters["identifier"] = req.Identifier
 	}
@@ -154,34 +174,13 @@ func (s *Service) Ship(ctx context.Context, id uint, req *dto.ExchangeShipReques
 	if order.Status != constants.ExchangeStatusPendingShip {
 		return nil, errors.New(errors.CodeExchangeStatusInvalid)
 	}
+	if !isShippingExchangeFlow(order.FlowType) {
+		return nil, errors.New(errors.CodeExchangeStatusInvalid, "该流程类型不支持发货")
+	}

-	newAsset, err := s.resolveAssetByIdentifier(ctx, order.OldAssetType, req.NewIdentifier)
-	if err != nil {
+	if err = s.shipWithTx(ctx, order, req); err != nil {
 		return nil, err
 	}
-	if newAsset.AssetType != order.OldAssetType {
-		return nil, errors.New(errors.CodeExchangeAssetTypeMismatch)
-	}
-	if newAsset.AssetStatus != 1 {
-		return nil, errors.New(errors.CodeExchangeNewAssetNotInStock)
-	}
-
-	updates := map[string]any{
-		"new_asset_type":       newAsset.AssetType,
-		"new_asset_id":         newAsset.AssetID,
-		"new_asset_identifier": newAsset.Identifier,
-		"express_company":      req.ExpressCompany,
-		"express_no":           req.ExpressNo,
-		"migrate_data":         req.MigrateData,
-		"updater":              middleware.GetUserIDFromContext(ctx),
-		"updated_at":           time.Now(),
-	}
-	if err = s.exchangeStore.UpdateStatus(ctx, id, constants.ExchangeStatusPendingShip, constants.ExchangeStatusShipped, updates); err != nil {
-		if err == gorm.ErrRecordNotFound {
-			return nil, errors.New(errors.CodeExchangeStatusInvalid)
-		}
-		return nil, errors.Wrap(errors.CodeDatabaseError, err, "更新换货单发货状态失败")
-	}

 	return s.Get(ctx, id)
 }
@@ -197,29 +196,23 @@ func (s *Service) Complete(ctx context.Context, id uint) error {
 	if order.Status != constants.ExchangeStatusShipped {
 		return errors.New(errors.CodeExchangeStatusInvalid)
 	}
-
-	updates := map[string]any{
-		"updater":    middleware.GetUserIDFromContext(ctx),
-		"updated_at": time.Now(),
+	if !isShippingExchangeFlow(order.FlowType) {
+		return errors.New(errors.CodeExchangeStatusInvalid, "该流程类型不支持确认完成")
 	}
-	if order.MigrateData {
-		var migrationBalance int64
-		migrationBalance, err = s.executeMigration(ctx, order)
-		if err != nil {
-			return err
+
+	return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		lockedOrder, lockErr := s.lockExchangeOrderByID(ctx, tx, id)
+		if lockErr != nil {
+			return lockErr
 		}
-		updates["migration_completed"] = true
-		updates["migration_balance"] = migrationBalance
-	}
-
-	if err = s.exchangeStore.UpdateStatus(ctx, id, constants.ExchangeStatusShipped, constants.ExchangeStatusCompleted, updates); err != nil {
-		if err == gorm.ErrRecordNotFound {
+		if !isShippingExchangeFlow(lockedOrder.FlowType) {
+			return errors.New(errors.CodeExchangeStatusInvalid, "该流程类型不支持确认完成")
+		}
+		if lockedOrder.Status != constants.ExchangeStatusShipped {
 			return errors.New(errors.CodeExchangeStatusInvalid)
 		}
-		return errors.Wrap(errors.CodeDatabaseError, err, "确认换货完成失败")
-	}
-
-	return nil
+		return s.completeExchangeWithTx(ctx, tx, lockedOrder, constants.ExchangeStatusShipped)
+	})
 }

 func (s *Service) Cancel(ctx context.Context, id uint, req *dto.ExchangeCancelRequest) error {
@@ -233,6 +226,9 @@ func (s *Service) Cancel(ctx context.Context, id uint, req *dto.ExchangeCancelRe
 	if order.Status != constants.ExchangeStatusPendingInfo && order.Status != constants.ExchangeStatusPendingShip {
 		return errors.New(errors.CodeExchangeStatusInvalid)
 	}
+	if !isShippingExchangeFlow(order.FlowType) {
+		return errors.New(errors.CodeExchangeStatusInvalid, "该流程类型不支持取消")
+	}

 	updates := map[string]any{
 		"updater":    middleware.GetUserIDFromContext(ctx),
@@ -271,13 +267,13 @@ func (s *Service) Renew(ctx context.Context, id uint) error {
 				}
 				return errors.Wrap(errors.CodeDatabaseError, err, "查询旧卡失败")
 			}
-			if card.AssetStatus != 3 {
+			if card.AssetStatus != constants.AssetStatusExchanged {
 				return errors.New(errors.CodeExchangeAssetNotExchanged)
 			}

 			if err = tx.Model(&model.IotCard{}).Where("id = ?", card.ID).Updates(map[string]any{
 				"generation":                         card.Generation + 1,
-				"asset_status":                       1,
+				"asset_status":                       constants.AssetStatusInStock,
 				"accumulated_recharge_by_series":     "{}",
 				"first_recharge_triggered_by_series": "{}",
 				"updater":                            middleware.GetUserIDFromContext(ctx),
@@ -286,8 +282,11 @@ func (s *Service) Renew(ctx context.Context, id uint) error {
 				return errors.Wrap(errors.CodeDatabaseError, err, "重置旧卡转新状态失败")
 			}

-			if err = tx.Where("virtual_no = ?", card.VirtualNo).Delete(&model.PersonalCustomerDevice{}).Error; err != nil {
-				return errors.Wrap(errors.CodeDatabaseError, err, "清理个人客户绑定失败")
+			cardKey := exchangeAssetBindingKey(&resolvedExchangeAsset{AssetType: constants.ExchangeAssetTypeIotCard, Card: &card, VirtualNo: card.VirtualNo})
+			if cardKey != "" {
+				if err = tx.Where("virtual_no = ?", cardKey).Delete(&model.PersonalCustomerDevice{}).Error; err != nil {
+					return errors.Wrap(errors.CodeDatabaseError, err, "清理个人客户绑定失败")
+				}
 			}

 			if err = tx.Where("resource_type = ? AND resource_id = ?", constants.ExchangeAssetTypeIotCard, card.ID).Delete(&model.AssetWallet{}).Error; err != nil {
@@ -298,7 +297,7 @@ func (s *Service) Renew(ctx context.Context, id uint) error {
 			if card.ShopID != nil {
 				shopTag = *card.ShopID
 			}
-			if err = tx.Create(&model.AssetWallet{ResourceType: constants.ExchangeAssetTypeIotCard, ResourceID: card.ID, Balance: 0, FrozenBalance: 0, Currency: "CNY", Status: 1, Version: 0, ShopIDTag: shopTag}).Error; err != nil {
+			if err = tx.Create(&model.AssetWallet{ResourceType: constants.ExchangeAssetTypeIotCard, ResourceID: card.ID, Balance: 0, FrozenBalance: 0, Currency: "CNY", Status: constants.AssetWalletStatusNormal, Version: 0, ShopIDTag: shopTag}).Error; err != nil {
 				return errors.Wrap(errors.CodeDatabaseError, err, "创建新钱包失败")
 			}
 			return nil
@@ -311,13 +310,13 @@ func (s *Service) Renew(ctx context.Context, id uint) error {
 			}
 			return errors.Wrap(errors.CodeDatabaseError, err, "查询旧设备失败")
 		}
-		if device.AssetStatus != 3 {
+		if device.AssetStatus != constants.AssetStatusExchanged {
 			return errors.New(errors.CodeExchangeAssetNotExchanged)
 		}

 		if err = tx.Model(&model.Device{}).Where("id = ?", device.ID).Updates(map[string]any{
 			"generation":                         device.Generation + 1,
-			"asset_status":                       1,
+			"asset_status":                       constants.AssetStatusInStock,
 			"accumulated_recharge_by_series":     "{}",
 			"first_recharge_triggered_by_series": "{}",
 			"updater":                            middleware.GetUserIDFromContext(ctx),
@@ -326,8 +325,11 @@ func (s *Service) Renew(ctx context.Context, id uint) error {
 			return errors.Wrap(errors.CodeDatabaseError, err, "重置旧设备转新状态失败")
 		}

-		if err = tx.Where("virtual_no = ?", device.VirtualNo).Delete(&model.PersonalCustomerDevice{}).Error; err != nil {
-			return errors.Wrap(errors.CodeDatabaseError, err, "清理个人客户绑定失败")
+		deviceKey := exchangeAssetBindingKey(&resolvedExchangeAsset{AssetType: constants.ExchangeAssetTypeDevice, Device: &device, VirtualNo: device.VirtualNo})
+		if deviceKey != "" {
+			if err = tx.Where("virtual_no = ?", deviceKey).Delete(&model.PersonalCustomerDevice{}).Error; err != nil {
+				return errors.Wrap(errors.CodeDatabaseError, err, "清理个人客户绑定失败")
+			}
 		}

 		if err = tx.Where("resource_type = ? AND resource_id = ?", constants.ExchangeAssetTypeDevice, device.ID).Delete(&model.AssetWallet{}).Error; err != nil {
@@ -338,7 +340,7 @@ func (s *Service) Renew(ctx context.Context, id uint) error {
 		if device.ShopID != nil {
 			shopTag = *device.ShopID
 		}
-		if err = tx.Create(&model.AssetWallet{ResourceType: constants.ExchangeAssetTypeDevice, ResourceID: device.ID, Balance: 0, FrozenBalance: 0, Currency: "CNY", Status: 1, Version: 0, ShopIDTag: shopTag}).Error; err != nil {
+		if err = tx.Create(&model.AssetWallet{ResourceType: constants.ExchangeAssetTypeDevice, ResourceID: device.ID, Balance: 0, FrozenBalance: 0, Currency: "CNY", Status: constants.AssetWalletStatusNormal, Version: 0, ShopIDTag: shopTag}).Error; err != nil {
 			return errors.Wrap(errors.CodeDatabaseError, err, "创建新钱包失败")
 		}
 		return nil
@@ -351,7 +353,11 @@ func (s *Service) GetPending(ctx context.Context, identifier string) (*dto.Clien
 		return nil, err
 	}

-	order, err := s.exchangeStore.FindActiveByOldAsset(ctx, asset.AssetType, asset.AssetID)
+	if !s.customerOwnsAsset(ctx, asset) {
+		return nil, errors.New(errors.CodeAssetNotFound)
+	}
+
+	order, err := s.exchangeStore.FindShippingPendingByOldAsset(ctx, asset.AssetType, asset.AssetID)
 	if err != nil {
 		if err == gorm.ErrRecordNotFound {
 			return nil, nil
@@ -362,14 +368,37 @@ func (s *Service) GetPending(ctx context.Context, identifier string) (*dto.Clien
 	return &dto.ClientExchangePendingResponse{
 		ID:             order.ID,
 		ExchangeNo:     order.ExchangeNo,
+		FlowType:       effectiveExchangeFlowType(order.FlowType),
 		Status:         order.Status,
-		StatusText:     exchangeStatusText(order.Status),
+		StatusName:     constants.GetExchangeStatusName(order.Status),
+		StatusText:     constants.GetExchangeStatusName(order.Status),
 		ExchangeReason: order.ExchangeReason,
 		CreatedAt:      order.CreatedAt,
 	}, nil
 }

 func (s *Service) SubmitShippingInfo(ctx context.Context, id uint, req *dto.ClientShippingInfoRequest) error {
+	order, err := s.exchangeStore.GetByID(ctx, id)
+	if err != nil {
+		if err == gorm.ErrRecordNotFound {
+			return errors.New(errors.CodeExchangeOrderNotFound)
+		}
+		return errors.Wrap(errors.CodeDatabaseError, err, "查询换货单失败")
+	}
+	if !isShippingExchangeFlow(order.FlowType) {
+		return errors.New(errors.CodeExchangeStatusInvalid, "该流程类型不支持填写收货信息")
+	}
+	if order.Status != constants.ExchangeStatusPendingInfo {
+		return errors.New(errors.CodeExchangeStatusInvalid)
+	}
+	oldAsset, err := s.resolveAssetByID(ctx, order.OldAssetType, order.OldAssetID)
+	if err != nil {
+		return err
+	}
+	if !s.customerOwnsAsset(ctx, oldAsset) {
+		return errors.New(errors.CodeExchangeOrderNotFound)
+	}
+
 	updates := map[string]any{
 		"recipient_name":    req.RecipientName,
 		"recipient_phone":   req.RecipientPhone,
@@ -411,8 +440,296 @@ func (s *Service) resolveAssetByIdentifier(ctx context.Context, expectedAssetTyp
 	}

 	if expectedAssetType == "" || expectedAssetType == constants.ExchangeAssetTypeIotCard {
+		card, err := s.iotCardStore.GetByIdentifier(ctx, identifier)
+		if err == nil {
+			if expectedAssetType != "" && expectedAssetType != constants.ExchangeAssetTypeIotCard {
+				return nil, errors.New(errors.CodeExchangeAssetTypeMismatch)
+			}
+			return &resolvedExchangeAsset{AssetType: constants.ExchangeAssetTypeIotCard, AssetID: card.ID, Identifier: identifier, VirtualNo: card.VirtualNo, AssetStatus: card.AssetStatus, ShopID: card.ShopID, Card: card}, nil
+		} else if err != gorm.ErrRecordNotFound {
+			return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询IoT卡失败")
+		}
+	}
+
+	return nil, errors.New(errors.CodeAssetNotFound)
+}
+
+func normalizeExchangeFlowType(flowType string) string {
+	flowType = strings.TrimSpace(flowType)
+	if flowType == "" {
+		return constants.ExchangeFlowTypeShipping
+	}
+	return flowType
+}
+
+func effectiveExchangeFlowType(flowType string) string {
+	return normalizeExchangeFlowType(flowType)
+}
+
+func isValidExchangeFlowType(flowType string) bool {
+	return flowType == constants.ExchangeFlowTypeShipping || flowType == constants.ExchangeFlowTypeDirect
+}
+
+func isShippingExchangeFlow(flowType string) bool {
+	return effectiveExchangeFlowType(flowType) == constants.ExchangeFlowTypeShipping
+}
+
+func (s *Service) createDirectExchange(ctx context.Context, req *dto.CreateExchangeRequest, oldAsset *resolvedExchangeAsset) (*dto.ExchangeOrderResponse, error) {
+	var orderID uint
+	migrateData := false
+	if req.MigrateData != nil {
+		migrateData = *req.MigrateData
+	}
+
+	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		lockedOldAsset, err := s.resolveAssetByIDWithTx(ctx, tx, oldAsset.AssetType, oldAsset.AssetID)
+		if err != nil {
+			return err
+		}
+		if lockedOldAsset.AssetStatus != constants.AssetStatusSold {
+			return errors.New(errors.CodeExchangeStatusInvalid, "旧资产状态不允许换货")
+		}
+		if err = s.ensureNoActiveExchangeWithTx(ctx, tx, lockedOldAsset.AssetType, lockedOldAsset.AssetID); err != nil {
+			return err
+		}
+
+		newAsset, err := s.resolveAssetByIdentifierWithTx(ctx, tx, "", req.NewIdentifier)
+		if err != nil {
+			return err
+		}
+		if newAsset.AssetType != lockedOldAsset.AssetType {
+			return errors.New(errors.CodeExchangeAssetTypeMismatch)
+		}
+
+		creator := middleware.GetUserIDFromContext(ctx)
+		order := &model.ExchangeOrder{
+			ExchangeNo:         model.GenerateExchangeNo(),
+			FlowType:           constants.ExchangeFlowTypeDirect,
+			OldAssetType:       lockedOldAsset.AssetType,
+			OldAssetID:         lockedOldAsset.AssetID,
+			OldAssetIdentifier: lockedOldAsset.Identifier,
+			NewAssetType:       newAsset.AssetType,
+			NewAssetID:         &newAsset.AssetID,
+			NewAssetIdentifier: newAsset.Identifier,
+			ExchangeReason:     req.ExchangeReason,
+			Remark:             req.Remark,
+			Status:             constants.ExchangeStatusPendingInfo,
+			MigrationCompleted: false,
+			MigrationBalance:   0,
+			MigrateData:        migrateData,
+			BaseModel:          model.BaseModel{Creator: creator, Updater: creator},
+		}
+		if lockedOldAsset.ShopID != nil {
+			order.ShopID = lockedOldAsset.ShopID
+		}
+		if err = tx.WithContext(ctx).Create(order).Error; err != nil {
+			return errors.Wrap(errors.CodeDatabaseError, err, "创建直接换货单失败")
+		}
+		if err = s.completeExchangeWithTx(ctx, tx, order, constants.ExchangeStatusPendingInfo); err != nil {
+			return err
+		}
+		orderID = order.ID
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return s.Get(ctx, orderID)
+}
+
+func (s *Service) shipWithTx(ctx context.Context, order *model.ExchangeOrder, req *dto.ExchangeShipRequest) error {
+	return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		lockedOrder, err := s.lockExchangeOrderByID(ctx, tx, order.ID)
+		if err != nil {
+			return err
+		}
+		if !isShippingExchangeFlow(lockedOrder.FlowType) {
+			return errors.New(errors.CodeExchangeStatusInvalid, "该流程类型不支持发货")
+		}
+		if lockedOrder.Status != constants.ExchangeStatusPendingShip {
+			return errors.New(errors.CodeExchangeStatusInvalid)
+		}
+
+		oldAsset, err := s.resolveAssetByIDWithTx(ctx, tx, lockedOrder.OldAssetType, lockedOrder.OldAssetID)
+		if err != nil {
+			return err
+		}
+		newAsset, err := s.resolveAssetByIdentifierWithTx(ctx, tx, "", req.NewIdentifier)
+		if err != nil {
+			return err
+		}
+		if newAsset.AssetType != lockedOrder.OldAssetType {
+			return errors.New(errors.CodeExchangeAssetTypeMismatch)
+		}
+		if err = s.validateExchangeAssetsWithTx(ctx, tx, lockedOrder.ID, oldAsset, newAsset); err != nil {
+			return err
+		}
+
+		now := time.Now()
+		result := tx.WithContext(ctx).Model(&model.ExchangeOrder{}).
+			Where("id = ? AND status = ?", lockedOrder.ID, constants.ExchangeStatusPendingShip).
+			Updates(map[string]any{
+				"new_asset_type":       newAsset.AssetType,
+				"new_asset_id":         newAsset.AssetID,
+				"new_asset_identifier": newAsset.Identifier,
+				"express_company":      req.ExpressCompany,
+				"express_no":           req.ExpressNo,
+				"migrate_data":         req.MigrateData,
+				"shipped_at":           now,
+				"status":               constants.ExchangeStatusShipped,
+				"updater":              middleware.GetUserIDFromContext(ctx),
+				"updated_at":           now,
+			})
+		if result.Error != nil {
+			return errors.Wrap(errors.CodeDatabaseError, result.Error, "更新换货单发货状态失败")
+		}
+		if result.RowsAffected == 0 {
+			return errors.New(errors.CodeExchangeStatusInvalid)
+		}
+		return nil
+	})
+}
+
+func (s *Service) completeExchangeWithTx(ctx context.Context, tx *gorm.DB, order *model.ExchangeOrder, fromStatus int) error {
+	if order.NewAssetID == nil || *order.NewAssetID == 0 || order.NewAssetIdentifier == "" {
+		return errors.New(errors.CodeInvalidParam, "新资产信息缺失")
+	}
+
+	oldAsset, err := s.resolveAssetByIDWithTx(ctx, tx, order.OldAssetType, order.OldAssetID)
+	if err != nil {
+		return err
+	}
+	newAsset, err := s.resolveAssetByIDWithTx(ctx, tx, order.NewAssetType, *order.NewAssetID)
+	if err != nil {
+		return err
+	}
+	if err = s.validateExchangeAssetsWithTx(ctx, tx, order.ID, oldAsset, newAsset); err != nil {
+		return err
+	}
+	if err = s.switchCustomerBindingWithTx(ctx, tx, oldAsset, newAsset); err != nil {
+		return err
+	}
+	if err = s.updateAssetStatusesForCompletion(ctx, tx, oldAsset, newAsset); err != nil {
+		return err
+	}
+
+	var migrationBalance int64
+	if order.MigrateData {
+		migrationBalance, err = s.executeMigrationWithTx(ctx, tx, order, oldAsset, newAsset)
+		if err != nil {
+			return err
+		}
+	}
+
+	now := time.Now()
+	updates := map[string]any{
+		"status":       constants.ExchangeStatusCompleted,
+		"completed_at": now,
+		"updater":      middleware.GetUserIDFromContext(ctx),
+		"updated_at":   now,
+	}
+	if order.MigrateData {
+		updates["migration_completed"] = true
+		updates["migration_balance"] = migrationBalance
+	}
+	result := tx.WithContext(ctx).Model(&model.ExchangeOrder{}).
+		Where("id = ? AND status = ?", order.ID, fromStatus).
+		Updates(updates)
+	if result.Error != nil {
+		return errors.Wrap(errors.CodeDatabaseError, result.Error, "确认换货完成失败")
+	}
+	if result.RowsAffected == 0 {
+		return errors.New(errors.CodeExchangeStatusInvalid)
+	}
+	return nil
+}
+
+func (s *Service) lockExchangeOrderByID(ctx context.Context, tx *gorm.DB, id uint) (*model.ExchangeOrder, error) {
+	var order model.ExchangeOrder
+	query := tx.WithContext(ctx).Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", id)
+	query = middleware.ApplyShopFilter(ctx, query)
+	if err := query.First(&order).Error; err != nil {
+		if err == gorm.ErrRecordNotFound {
+			return nil, errors.New(errors.CodeExchangeOrderNotFound)
+		}
+		return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询换货单失败")
+	}
+	return &order, nil
+}
+
+func (s *Service) resolveAssetByID(ctx context.Context, assetType string, assetID uint) (*resolvedExchangeAsset, error) {
+	if assetType == constants.ExchangeAssetTypeIotCard {
+		card, err := s.iotCardStore.GetByID(ctx, assetID)
+		if err != nil {
+			if err == gorm.ErrRecordNotFound {
+				return nil, errors.New(errors.CodeAssetNotFound)
+			}
+			return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询IoT卡失败")
+		}
+		return &resolvedExchangeAsset{AssetType: constants.ExchangeAssetTypeIotCard, AssetID: card.ID, Identifier: card.VirtualNo, VirtualNo: card.VirtualNo, AssetStatus: card.AssetStatus, ShopID: card.ShopID, Card: card}, nil
+	}
+	if assetType == constants.ExchangeAssetTypeDevice {
+		device, err := s.deviceStore.GetByID(ctx, assetID)
+		if err != nil {
+			if err == gorm.ErrRecordNotFound {
+				return nil, errors.New(errors.CodeAssetNotFound)
+			}
+			return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询设备失败")
+		}
+		return &resolvedExchangeAsset{AssetType: constants.ExchangeAssetTypeDevice, AssetID: device.ID, Identifier: preferredDeviceIdentifier(device), VirtualNo: device.VirtualNo, AssetStatus: device.AssetStatus, ShopID: device.ShopID, Device: device}, nil
+	}
+	return nil, errors.New(errors.CodeInvalidParam, "资产类型不合法")
+}
+
+func (s *Service) resolveAssetByIDWithTx(ctx context.Context, tx *gorm.DB, assetType string, assetID uint) (*resolvedExchangeAsset, error) {
+	if assetType == constants.ExchangeAssetTypeIotCard {
 		var card model.IotCard
-		query := s.db.WithContext(ctx).Where("virtual_no = ? OR iccid = ? OR msisdn = ?", identifier, identifier, identifier)
+		query := tx.WithContext(ctx).Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", assetID)
+		query = middleware.ApplyShopFilter(ctx, query)
+		if err := query.First(&card).Error; err != nil {
+			if err == gorm.ErrRecordNotFound {
+				return nil, errors.New(errors.CodeAssetNotFound)
+			}
+			return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询IoT卡失败")
+		}
+		return &resolvedExchangeAsset{AssetType: constants.ExchangeAssetTypeIotCard, AssetID: card.ID, Identifier: card.VirtualNo, VirtualNo: card.VirtualNo, AssetStatus: card.AssetStatus, ShopID: card.ShopID, Card: &card}, nil
+	}
+	if assetType == constants.ExchangeAssetTypeDevice {
+		var device model.Device
+		query := tx.WithContext(ctx).Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", assetID)
+		query = middleware.ApplyShopFilter(ctx, query)
+		if err := query.First(&device).Error; err != nil {
+			if err == gorm.ErrRecordNotFound {
+				return nil, errors.New(errors.CodeAssetNotFound)
+			}
+			return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询设备失败")
+		}
+		return &resolvedExchangeAsset{AssetType: constants.ExchangeAssetTypeDevice, AssetID: device.ID, Identifier: preferredDeviceIdentifier(&device), VirtualNo: device.VirtualNo, AssetStatus: device.AssetStatus, ShopID: device.ShopID, Device: &device}, nil
+	}
+	return nil, errors.New(errors.CodeInvalidParam, "资产类型不合法")
+}
+
+func (s *Service) resolveAssetByIdentifierWithTx(ctx context.Context, tx *gorm.DB, expectedAssetType, identifier string) (*resolvedExchangeAsset, error) {
+	if expectedAssetType == "" || expectedAssetType == constants.ExchangeAssetTypeDevice {
+		var device model.Device
+		query := tx.WithContext(ctx).
+			Clauses(clause.Locking{Strength: "UPDATE"}).
+			Where("virtual_no = ? OR imei = ? OR sn = ?", identifier, identifier, identifier)
+		query = middleware.ApplyShopFilter(ctx, query)
+		if err := query.First(&device).Error; err == nil {
+			if expectedAssetType != "" && expectedAssetType != constants.ExchangeAssetTypeDevice {
+				return nil, errors.New(errors.CodeExchangeAssetTypeMismatch)
+			}
+			return &resolvedExchangeAsset{AssetType: constants.ExchangeAssetTypeDevice, AssetID: device.ID, Identifier: identifier, VirtualNo: device.VirtualNo, AssetStatus: device.AssetStatus, ShopID: device.ShopID, Device: &device}, nil
+		} else if err != gorm.ErrRecordNotFound {
+			return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询设备失败")
+		}
+	}
+	if expectedAssetType == "" || expectedAssetType == constants.ExchangeAssetTypeIotCard {
+		var card model.IotCard
+		query := tx.WithContext(ctx).
+			Clauses(clause.Locking{Strength: "UPDATE"}).
+			Where("virtual_no = ? OR iccid = ? OR msisdn = ? OR iccid_19 = ? OR iccid_20 = ?", identifier, identifier, identifier, identifier, identifier)
 		query = middleware.ApplyShopFilter(ctx, query)
 		if err := query.First(&card).Error; err == nil {
 			if expectedAssetType != "" && expectedAssetType != constants.ExchangeAssetTypeIotCard {
@@ -423,10 +740,213 @@ func (s *Service) resolveAssetByIdentifier(ctx context.Context, expectedAssetTyp
 			return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询IoT卡失败")
 		}
 	}
-
 	return nil, errors.New(errors.CodeAssetNotFound)
 }

+func (s *Service) ensureNoActiveExchangeWithTx(ctx context.Context, tx *gorm.DB, assetType string, assetID uint) error {
+	var count int64
+	query := tx.WithContext(ctx).Model(&model.ExchangeOrder{}).
+		Where("old_asset_type = ? AND old_asset_id = ?", assetType, assetID).
+		Where("status IN ?", []int{constants.ExchangeStatusPendingInfo, constants.ExchangeStatusPendingShip, constants.ExchangeStatusShipped})
+	query = middleware.ApplyShopFilter(ctx, query)
+	if err := query.Count(&count).Error; err != nil {
+		return errors.Wrap(errors.CodeDatabaseError, err, "查询进行中换货单失败")
+	}
+	if count > 0 {
+		return errors.New(errors.CodeExchangeInProgress)
+	}
+	return nil
+}
+
+func (s *Service) validateExchangeAssetsWithTx(ctx context.Context, tx *gorm.DB, orderID uint, oldAsset, newAsset *resolvedExchangeAsset) error {
+	if oldAsset.AssetType != newAsset.AssetType {
+		return errors.New(errors.CodeExchangeAssetTypeMismatch)
+	}
+	if oldAsset.AssetStatus != constants.AssetStatusSold {
+		return errors.New(errors.CodeExchangeStatusInvalid, "旧资产状态不允许换货")
+	}
+	if newAsset.AssetStatus != constants.AssetStatusInStock {
+		return errors.New(errors.CodeExchangeNewAssetNotInStock)
+	}
+	if !sameShopID(oldAsset.ShopID, newAsset.ShopID) {
+		return errors.New(errors.CodeForbidden, "新旧资产归属不一致")
+	}
+	if err := s.ensureNewAssetBindingAvailableWithTx(ctx, tx, oldAsset, newAsset); err != nil {
+		return err
+	}
+	occupied, err := s.hasShippingNewAssetOccupiedWithTx(ctx, tx, newAsset.AssetType, newAsset.AssetID, orderID)
+	if err != nil {
+		return err
+	}
+	if occupied {
+		return errors.New(errors.CodeExchangeStatusInvalid, "新资产已被其他换货单占用")
+	}
+	return nil
+}
+
+func (s *Service) ensureNewAssetBindingAvailableWithTx(ctx context.Context, tx *gorm.DB, oldAsset, newAsset *resolvedExchangeAsset) error {
+	oldKey := exchangeAssetBindingKey(oldAsset)
+	newKey := exchangeAssetBindingKey(newAsset)
+	oldBindCount := int64(0)
+	if oldKey != "" {
+		if err := tx.WithContext(ctx).Model(&model.PersonalCustomerDevice{}).
+			Where("virtual_no = ? AND status = ?", oldKey, constants.StatusEnabled).
+			Count(&oldBindCount).Error; err != nil {
+			return errors.Wrap(errors.CodeDatabaseError, err, "查询旧资产客户绑定失败")
+		}
+	}
+	if oldBindCount > 0 && newKey == "" {
+		return errors.New(errors.CodeExchangeStatusInvalid, "新资产无法承接客户绑定")
+	}
+	if newKey == "" {
+		return nil
+	}
+	var newBindCount int64
+	if err := tx.WithContext(ctx).Model(&model.PersonalCustomerDevice{}).
+		Where("virtual_no = ? AND status = ?", newKey, constants.StatusEnabled).
+		Count(&newBindCount).Error; err != nil {
+		return errors.Wrap(errors.CodeDatabaseError, err, "查询新资产客户绑定失败")
+	}
+	if newBindCount > 0 {
+		return errors.New(errors.CodeExchangeStatusInvalid, "新资产已存在有效客户绑定")
+	}
+	return nil
+}
+
+func (s *Service) switchCustomerBindingWithTx(ctx context.Context, tx *gorm.DB, oldAsset, newAsset *resolvedExchangeAsset) error {
+	oldKey := exchangeAssetBindingKey(oldAsset)
+	if oldKey == "" {
+		return nil
+	}
+	newKey := exchangeAssetBindingKey(newAsset)
+	if newKey == "" {
+		return errors.New(errors.CodeExchangeStatusInvalid, "新资产无法承接客户绑定")
+	}
+	if err := tx.WithContext(ctx).Model(&model.PersonalCustomerDevice{}).
+		Where("virtual_no = ? AND status = ?", oldKey, constants.StatusEnabled).
+		Updates(map[string]any{"virtual_no": newKey, "updated_at": time.Now()}).Error; err != nil {
+		return errors.Wrap(errors.CodeDatabaseError, err, "更新客户绑定关系失败")
+	}
+	return nil
+}
+
+func (s *Service) updateAssetStatusesForCompletion(ctx context.Context, tx *gorm.DB, oldAsset, newAsset *resolvedExchangeAsset) error {
+	now := time.Now()
+	if oldAsset.AssetType == constants.ExchangeAssetTypeIotCard {
+		result := tx.WithContext(ctx).Model(&model.IotCard{}).
+			Where("id = ? AND asset_status = ?", oldAsset.AssetID, constants.AssetStatusSold).
+			Updates(map[string]any{"asset_status": constants.AssetStatusExchanged, "updated_at": now})
+		if result.Error != nil {
+			return errors.Wrap(errors.CodeDatabaseError, result.Error, "更新旧卡状态失败")
+		}
+		if result.RowsAffected == 0 {
+			return errors.New(errors.CodeExchangeStatusInvalid, "旧资产状态不允许换货")
+		}
+		result = tx.WithContext(ctx).Model(&model.IotCard{}).
+			Where("id = ? AND asset_status = ?", newAsset.AssetID, constants.AssetStatusInStock).
+			Updates(map[string]any{"asset_status": constants.AssetStatusSold, "updated_at": now})
+		if result.Error != nil {
+			return errors.Wrap(errors.CodeDatabaseError, result.Error, "更新新卡状态失败")
+		}
+		if result.RowsAffected == 0 {
+			return errors.New(errors.CodeExchangeNewAssetNotInStock)
+		}
+		return nil
+	}
+	result := tx.WithContext(ctx).Model(&model.Device{}).
+		Where("id = ? AND asset_status = ?", oldAsset.AssetID, constants.AssetStatusSold).
+		Updates(map[string]any{"asset_status": constants.AssetStatusExchanged, "updated_at": now})
+	if result.Error != nil {
+		return errors.Wrap(errors.CodeDatabaseError, result.Error, "更新旧设备状态失败")
+	}
+	if result.RowsAffected == 0 {
+		return errors.New(errors.CodeExchangeStatusInvalid, "旧资产状态不允许换货")
+	}
+	result = tx.WithContext(ctx).Model(&model.Device{}).
+		Where("id = ? AND asset_status = ?", newAsset.AssetID, constants.AssetStatusInStock).
+		Updates(map[string]any{"asset_status": constants.AssetStatusSold, "updated_at": now})
+	if result.Error != nil {
+		return errors.Wrap(errors.CodeDatabaseError, result.Error, "更新新设备状态失败")
+	}
+	if result.RowsAffected == 0 {
+		return errors.New(errors.CodeExchangeNewAssetNotInStock)
+	}
+	return nil
+}
+
+func (s *Service) hasShippingNewAssetOccupiedWithTx(ctx context.Context, tx *gorm.DB, assetType string, assetID uint, excludeOrderID uint) (bool, error) {
+	var count int64
+	query := tx.WithContext(ctx).Model(&model.ExchangeOrder{}).
+		Where("new_asset_type = ? AND new_asset_id = ?", assetType, assetID).
+		Where("status = ?", constants.ExchangeStatusShipped).
+		Where("COALESCE(NULLIF(flow_type, ''), ?) = ?", constants.ExchangeFlowTypeShipping, constants.ExchangeFlowTypeShipping)
+	if excludeOrderID > 0 {
+		query = query.Where("id <> ?", excludeOrderID)
+	}
+	query = middleware.ApplyShopFilter(ctx, query)
+	if err := query.Count(&count).Error; err != nil {
+		return false, errors.Wrap(errors.CodeDatabaseError, err, "查询新资产占用失败")
+	}
+	return count > 0, nil
+}
+
+func (s *Service) customerOwnsAsset(ctx context.Context, asset *resolvedExchangeAsset) bool {
+	customerID := middleware.GetCustomerIDFromContext(ctx)
+	if customerID == 0 || asset == nil {
+		return false
+	}
+	key := exchangeAssetBindingKey(asset)
+	if key == "" {
+		return false
+	}
+	var count int64
+	if err := s.db.WithContext(ctx).Model(&model.PersonalCustomerDevice{}).
+		Where("customer_id = ? AND virtual_no = ? AND status = ?", customerID, key, constants.StatusEnabled).
+		Count(&count).Error; err != nil {
+		return false
+	}
+	return count > 0
+}
+
+func exchangeAssetBindingKey(asset *resolvedExchangeAsset) string {
+	if asset == nil {
+		return ""
+	}
+	if asset.AssetType == constants.ExchangeAssetTypeIotCard {
+		if asset.Card != nil {
+			return asset.Card.VirtualNo
+		}
+		return asset.VirtualNo
+	}
+	if asset.Device == nil {
+		return asset.VirtualNo
+	}
+	if asset.Device.VirtualNo != "" {
+		return asset.Device.VirtualNo
+	}
+	return asset.Device.IMEI
+}
+
+func preferredDeviceIdentifier(device *model.Device) string {
+	if device == nil {
+		return ""
+	}
+	if device.VirtualNo != "" {
+		return device.VirtualNo
+	}
+	if device.IMEI != "" {
+		return device.IMEI
+	}
+	return device.SN
+}
+
+func sameShopID(left, right *uint) bool {
+	if left == nil || right == nil {
+		return left == nil && right == nil
+	}
+	return *left == *right
+}
+
 func (s *Service) toExchangeOrderResponse(order *model.ExchangeOrder) *dto.ExchangeOrderResponse {
 	if order == nil {
 		return nil
@@ -438,6 +958,8 @@ func (s *Service) toExchangeOrderResponse(order *model.ExchangeOrder) *dto.Excha
 	return &dto.ExchangeOrderResponse{
 		ID:                 order.ID,
 		ExchangeNo:         order.ExchangeNo,
+		FlowType:           effectiveExchangeFlowType(order.FlowType),
+		FlowTypeName:       constants.GetExchangeFlowTypeName(order.FlowType),
 		OldAssetType:       order.OldAssetType,
 		OldAssetID:         order.OldAssetID,
 		OldAssetIdentifier: order.OldAssetIdentifier,
@@ -452,10 +974,13 @@ func (s *Service) toExchangeOrderResponse(order *model.ExchangeOrder) *dto.Excha
 		MigrateData:        order.MigrateData,
 		MigrationCompleted: order.MigrationCompleted,
 		MigrationBalance:   order.MigrationBalance,
+		ShippedAt:          order.ShippedAt,
+		CompletedAt:        order.CompletedAt,
 		ExchangeReason:     order.ExchangeReason,
 		Remark:             order.Remark,
 		Status:             order.Status,
-		StatusText:         exchangeStatusText(order.Status),
+		StatusName:         constants.GetExchangeStatusName(order.Status),
+		StatusText:         constants.GetExchangeStatusName(order.Status),
 		ShopID:             order.ShopID,
 		CreatedAt:          order.CreatedAt,
 		UpdatedAt:          order.UpdatedAt,
@@ -464,20 +989,3 @@ func (s *Service) toExchangeOrderResponse(order *model.ExchangeOrder) *dto.Excha
 		Updater:            order.Updater,
 	}
 }
-
-func exchangeStatusText(status int) string {
-	switch status {
-	case constants.ExchangeStatusPendingInfo:
-		return "待填写信息"
-	case constants.ExchangeStatusPendingShip:
-		return "待发货"
-	case constants.ExchangeStatusShipped:
-		return "已发货待确认"
-	case constants.ExchangeStatusCompleted:
-		return "已完成"
-	case constants.ExchangeStatusCancelled:
-		return "已取消"
-	default:
-		return "未知状态"
-	}
-}
--- a/internal/store/postgres/exchange_order_store.go
+++ b/internal/store/postgres/exchange_order_store.go
@@ -43,6 +43,9 @@ func (s *ExchangeOrderStore) List(ctx context.Context, filters map[string]any, p
 	if status, ok := filters["status"].(int); ok && status > 0 {
 		query = query.Where("status = ?", status)
 	}
+	if flowType, ok := filters["flow_type"].(string); ok && flowType != "" {
+		query = query.Where("COALESCE(NULLIF(flow_type, ''), ?) = ?", constants.ExchangeFlowTypeShipping, flowType)
+	}
 	if identifier, ok := filters["identifier"].(string); ok && identifier != "" {
 		like := "%" + identifier + "%"
 		query = query.Where("old_asset_identifier LIKE ? OR new_asset_identifier LIKE ?", like, like)
@@ -93,13 +96,14 @@ func (s *ExchangeOrderStore) UpdateStatus(ctx context.Context, id uint, fromStat
 	return nil
 }

-// FindByNewAssetID 通过新资产类型和ID查询换货来源记录
-// 用于换货链追溯：找到将当前资产作为换货结果的换货单，从而得到前代资产信息
+// FindByNewAssetID 通过新资产类型和ID查询已完成换货来源记录
+// 用于换货链追溯：找到将当前资产作为换货结果的已完成换货单，从而得到前代资产信息
 // 若未找到则返回 nil, nil 表示该资产无前代
 func (s *ExchangeOrderStore) FindByNewAssetID(ctx context.Context, assetType string, assetID uint) (*model.ExchangeOrder, error) {
 	var order model.ExchangeOrder
 	err := s.db.WithContext(ctx).
 		Where("new_asset_id = ? AND new_asset_type = ?", assetID, assetType).
+		Where("status = ?", constants.ExchangeStatusCompleted).
 		Order("id DESC").
 		First(&order).Error
 	if err != nil {
@@ -111,6 +115,21 @@ func (s *ExchangeOrderStore) FindByNewAssetID(ctx context.Context, assetType str
 	return &order, nil
 }

+// FindShippingPendingByOldAsset 查询客户端可见的物流换货待处理单
+// 历史 flow_type 为空时按 shipping 兼容处理。
+func (s *ExchangeOrderStore) FindShippingPendingByOldAsset(ctx context.Context, assetType string, assetID uint) (*model.ExchangeOrder, error) {
+	var order model.ExchangeOrder
+	query := s.db.WithContext(ctx).
+		Where("old_asset_type = ? AND old_asset_id = ?", assetType, assetID).
+		Where("status IN ?", []int{constants.ExchangeStatusPendingInfo, constants.ExchangeStatusPendingShip, constants.ExchangeStatusShipped}).
+		Where("COALESCE(NULLIF(flow_type, ''), ?) = ?", constants.ExchangeFlowTypeShipping, constants.ExchangeFlowTypeShipping)
+	query = middleware.ApplyShopFilter(ctx, query)
+	if err := query.Order("id DESC").First(&order).Error; err != nil {
+		return nil, err
+	}
+	return &order, nil
+}
+
 func (s *ExchangeOrderStore) FindActiveByOldAsset(ctx context.Context, assetType string, assetID uint) (*model.ExchangeOrder, error) {
 	var order model.ExchangeOrder
 	query := s.db.WithContext(ctx).
--- a/migrations/000152_add_direct_exchange_flow.down.sql
+++ b/migrations/000152_add_direct_exchange_flow.down.sql
@@ -0,0 +1,9 @@
+-- 回滚直接换货流程字段与索引
+DROP INDEX IF EXISTS idx_exchange_order_completed_at;
+DROP INDEX IF EXISTS idx_exchange_order_new_asset;
+DROP INDEX IF EXISTS idx_exchange_order_flow_status;
+
+ALTER TABLE tb_exchange_order
+    DROP COLUMN IF EXISTS completed_at,
+    DROP COLUMN IF EXISTS shipped_at,
+    DROP COLUMN IF EXISTS flow_type;
--- a/migrations/000152_add_direct_exchange_flow.up.sql
+++ b/migrations/000152_add_direct_exchange_flow.up.sql
@@ -0,0 +1,21 @@
+-- 为换货单补充流程类型和真实业务时间字段
+ALTER TABLE tb_exchange_order
+    ADD COLUMN IF NOT EXISTS flow_type VARCHAR(20) NOT NULL DEFAULT 'shipping',
+    ADD COLUMN IF NOT EXISTS shipped_at TIMESTAMPTZ,
+    ADD COLUMN IF NOT EXISTS completed_at TIMESTAMPTZ;
+
+COMMENT ON COLUMN tb_exchange_order.flow_type IS '换货流程类型 shipping-物流换货 direct-直接换货';
+COMMENT ON COLUMN tb_exchange_order.shipped_at IS '发货时间，仅物流换货发货成功后写入';
+COMMENT ON COLUMN tb_exchange_order.completed_at IS '换货完成时间，物流换货和直接换货完成后写入';
+
+CREATE INDEX IF NOT EXISTS idx_exchange_order_flow_status
+    ON tb_exchange_order (flow_type, status)
+    WHERE deleted_at IS NULL;
+
+CREATE INDEX IF NOT EXISTS idx_exchange_order_new_asset
+    ON tb_exchange_order (new_asset_type, new_asset_id)
+    WHERE deleted_at IS NULL AND new_asset_id IS NOT NULL;
+
+CREATE INDEX IF NOT EXISTS idx_exchange_order_completed_at
+    ON tb_exchange_order (completed_at DESC)
+    WHERE deleted_at IS NULL AND completed_at IS NOT NULL;
--- a/pkg/constants/constants.go
+++ b/pkg/constants/constants.go
@@ -386,6 +386,12 @@ const (
 	ExchangeStatusCancelled   = 5 // 已取消
 )

+// 换货流程类型常量
+const (
+	ExchangeFlowTypeShipping = "shipping" // 物流换货
+	ExchangeFlowTypeDirect   = "direct"   // 直接换货
+)
+
 // 换货资产类型常量
 const (
 	ExchangeAssetTypeIotCard = "iot_card" // 物联网卡
@@ -472,3 +478,15 @@ func GetExchangeStatusName(status int) string {
 		return "未知"
 	}
 }
+
+// GetExchangeFlowTypeName 获取换货流程类型名称
+func GetExchangeFlowTypeName(flowType string) string {
+	switch flowType {
+	case "", ExchangeFlowTypeShipping:
+		return "物流换货"
+	case ExchangeFlowTypeDirect:
+		return "直接换货"
+	default:
+		return "未知"
+	}
+}
--- a/pkg/constants/wallet.go
+++ b/pkg/constants/wallet.go
@@ -68,6 +68,7 @@ const (
 	AssetTransactionTypeRecharge = "recharge" // 充值
 	AssetTransactionTypeDeduct   = "deduct"   // 扣款
 	AssetTransactionTypeRefund   = "refund"   // 退款
+	AssetTransactionTypeExchange = "exchange" // 换货迁移
 )

 // 资产充值订单号前缀
@@ -114,6 +115,7 @@ const (
 	ReferenceTypeWithdrawal = "withdrawal" // 提现
 	ReferenceTypeTopup      = "topup"      // 充值
 	ReferenceTypeRefund     = "refund"     // 退款
+	ReferenceTypeExchange   = "exchange"   // 换货
 )

 // ========== Redis Key 生成函数 ==========