From 3407e30395886016d0df1866adfe283d603f5f5e Mon Sep 17 00:00:00 2001 From: zerosaturation Date: Thu, 23 Jul 2026 12:28:57 +0800 Subject: [PATCH] refactor(assetService): CreateMintOrder 3-phase txn + orphan reconciliation (P0-2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - CreateMintOrder 拆三段: txn1(PENDING→PROCESSING,不调RPC/不写asset) → 事务外 UpdateCrystalBalance 扣水晶 → txn2(建asset+registry→SUCCESS); 消除 DB 事务内嵌跨服务 gRPC(连接池占用+跨服务事务风险)。 - 任一步失败 markMintOrderFailed 独立事务标 FAILED(不回滚已扣水晶=审计流水, 重试靠 Task1 source_id 幂等 + Task2 入口短路防双扣); nil-cause 防御。 - 孤儿订单对账 ReconcileStuckMintOrders: 扫陈旧 PROCESSING 单,查 mint_cost 流水— 已扣未建→FOR UPDATE 行锁下幂等补完 SUCCESS(串行化并发防双建),未扣→FAILED,绝不退款。 - main.go 接线周期 worker(MINT_RECONCILE_INTERVAL_SEC 默认600s,0关,优雅退出)。 - 测试: RPC失败落FAILED / 对账未扣→FAILED / 已扣未建→补完(含幂等二次跑)。 Co-Authored-By: Claude --- backend/services/assetService/main.go | 59 ++ .../assetService/service/mint_service.go | 583 +++++++++++++++--- .../service/mint_service_idempotency_test.go | 263 +++++++- .../service/peripheral_test_helpers_test.go | 5 + 4 files changed, 827 insertions(+), 83 deletions(-) diff --git a/backend/services/assetService/main.go b/backend/services/assetService/main.go index f43de14..1be6bd3 100644 --- a/backend/services/assetService/main.go +++ b/backend/services/assetService/main.go @@ -1,11 +1,13 @@ package main import ( + "context" "flag" "fmt" "os" "os/signal" "strconv" + "sync" "syscall" "time" @@ -32,6 +34,8 @@ import ( starbookRepo "github.com/topfans/backend/services/starbookService/repository" "github.com/topfans/backend/services/assetService/util" "github.com/topfans/backend/services/assetService/util/ossutil" + + "go.uber.org/zap" ) var ( @@ -43,6 +47,9 @@ var ( dbName = flag.String("db-name", getEnv("DB_NAME", "top-fans"), "Database name") userServiceURL = flag.String("user-service-url", getEnv("USER_SERVICE_URL", "tri://localhost:20000"), "User service URL") healthHandler *health.Handler + // ★ mint-task-3 review P1-1: 对账 worker 周期(默认 10 分钟;设为 0 = 关闭) + reconcileIntervalSec = flag.Int("reconcile-interval-sec", getEnvInt("MINT_RECONCILE_INTERVAL_SEC", 600), "Reconcile stuck mint orders interval (seconds); 0 disables") + reconcileStaleMin = flag.Int("reconcile-stale-minutes", getEnvInt("MINT_RECONCILE_STALE_MINUTES", 10), "Stale threshold for stuck PROCESSING orders (minutes)") ) func getEnv(key, fallback string) string { @@ -200,6 +207,52 @@ func main() { }() logger.Logger.Info("Season reset worker started") + // ★ mint-task-3 review P1-1: 对账 worker + // - 周期触发 ReconcileStuckMintOrders(扫描陈旧 PROCESSING 订单并补单/标 FAILED) + // - 间隔与陈旧阈值通过 flag/env 可调(默认 10 分钟,生产可放大到 30+ 分钟) + // - interval=0 时关闭(本地/压测) + // - 随服务优雅退出:cancelFunc + WaitGroup + var reconcileWG sync.WaitGroup + var reconcileCancel context.CancelFunc + if *reconcileIntervalSec > 0 { + interval := time.Duration(*reconcileIntervalSec) * time.Second + stale := time.Duration(*reconcileStaleMin) * time.Minute + reconcileCtx, cancel := context.WithCancel(context.Background()) + reconcileCancel = cancel + reconcileWG.Add(1) + go func() { + defer reconcileWG.Done() + // 启动后先做一次"热启动对账",把上次服务崩溃遗留的 PROCESSING 订单先处理一波 + // (避免冷启动后还要等满 interval 才有第一波对账) + logger.Logger.Info("Mint reconcile worker bootstrapping", + zap.Duration("interval", interval), + zap.Duration("stale_after", stale)) + if err := mintService.ReconcileStuckMintOrders(reconcileCtx, stale); err != nil { + logger.Logger.Error("reconcile bootstrap failed", zap.Error(err)) + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-reconcileCtx.Done(): + logger.Logger.Info("Mint reconcile worker exiting") + return + case <-ticker.C: + // 每轮对账本身已有 INFO 日志(scanned/recovered/marked_failed/skipped), + // 顶层不需要再包一行 INFO。 + if err := mintService.ReconcileStuckMintOrders(reconcileCtx, stale); err != nil { + logger.Logger.Error("reconcile tick failed", zap.Error(err)) + } + } + } + }() + logger.Logger.Info("Mint reconcile worker started", + zap.Duration("interval", interval), + zap.Duration("stale_after", stale)) + } else { + logger.Logger.Info("Mint reconcile worker disabled (interval=0)") + } + // 创建 Dubbo 服务器 srv, err := server.NewServer( server.WithServerProtocol( @@ -244,6 +297,12 @@ func main() { if healthHandler != nil { healthHandler.Stop() } + + // 停止 mint 对账 worker(若已启动) + if reconcileCancel != nil { + reconcileCancel() + reconcileWG.Wait() + } } // autoMigrate 自动迁移数据库表 diff --git a/backend/services/assetService/service/mint_service.go b/backend/services/assetService/service/mint_service.go index 23c987c..e8423f0 100644 --- a/backend/services/assetService/service/mint_service.go +++ b/backend/services/assetService/service/mint_service.go @@ -3,6 +3,7 @@ package service import ( "context" "crypto/rand" + "errors" "fmt" "math/big" "net/url" @@ -29,6 +30,7 @@ import ( starbookRepo "github.com/topfans/backend/services/starbookService/repository" "go.uber.org/zap" "gorm.io/gorm" + "gorm.io/gorm/clause" "google.golang.org/grpc/codes" ) @@ -60,6 +62,11 @@ type MintService interface { // UpdateMintCountAndBoost 更新铸爱次数和收益提升 UpdateMintCountAndBoost(ctx context.Context, tx *gorm.DB, userID, starID int64, boostBps int32) error + + // ReconcileStuckMintOrders 对账:扫描 status=PROCESSING 且 updated_at + // 早于 staleAfter 的订单,根据是否已扣水晶做"幂等补单"或"安全标 FAILED"。 + // 调用方:可由 worker 周期触发,也可手动调用(管理后台/运维)。 + ReconcileStuckMintOrders(ctx context.Context, staleAfter time.Duration) error } // mintService 铸造服务实现 @@ -207,6 +214,20 @@ func (s *mintService) PreCreateMintOrder(req *pb.PreCreateMintOrderRequest, user } // CreateMintOrder 创建铸造订单 +// +// ★ mint-task-3 P0 重构:把 RPC 移出 DB 事务,消除连接池占用 + 跨服务事务风险, +// 并引入 PROCESSING 中间态,确保任何阶段失败都不会留下"状态机不对齐"的订单。 +// +// 三段式事务: +// - txn_local#1: PENDING → PROCESSING + 校验 + cost 计算 + mint_count 自增 + 序列同步 +// (不调 RPC、不写 asset) +// - 事务外 RPC: UpdateCrystalBalance 扣水晶(靠 mint-task-1 source_id 幂等保护) +// - txn_local#2: 建 asset/registry + 订单 → SUCCESS +// +// 失败补偿:任一阶段失败 → 独立事务 markMintOrderFailed(不回调水晶,不重试)。 +// +// 状态机:PENDING → PROCESSING → SUCCESS | FAILED。FAILED 订单不再二次扣, +// 用户必须 PreCreateMintOrder 新建订单才能重试。 func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, starID int64) (*pb.CreateMintOrderResponse, error) { // 1. 参数验证 if !validator.ValidateUserID(userID) { @@ -267,28 +288,36 @@ func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, st currentMintCount = 0 } - // 3. 使用事务创建铸造订单（或将阶段一订单推进到 PROCESSING） + // ==================================================================== + // 阶段1 (txn_local#1):PENDING → PROCESSING + 校验 + 费用/序列/mint_count + // - 不调 RPC + // - 不写 asset + // - 失败回滚到 PENDING(订单本就是 PENDING,无副作用) + // ==================================================================== + var costCrystal int64 var mintOrder *models.MintOrder - var asset *models.Asset - var newBalance int64 - var capturedCostCrystal int64 // 捕获铸造消耗 + var localMintCost *models.MintCostConfig + var boostBps int32 + err = s.db.Transaction(func(tx *gorm.DB) error { - // 用局部变量捕获事务内获取的 newBalance - var capturedBalance int64 - defer func() { newBalance = capturedBalance }() // 3.0 取出阶段一订单，并校验状态/所有者 - logger.Logger.Info("[MintOrder] Step 3.0: 获取订单", zap.String("order_id", req.OrderId), zap.Int64("user_id", userID), zap.Int64("star_id", starID)) + logger.Logger.Info("[MintOrder] Step 3.0: 获取订单", + zap.String("order_id", req.OrderId), + zap.Int64("user_id", userID), + zap.Int64("star_id", starID)) existing, err := s.mintOrderRepo.GetByOrderIDAndUser(req.OrderId, userID, starID) if err != nil { - logger.Logger.Error("[MintOrder] Step 3.0 失败: 获取订单错误", zap.String("order_id", req.OrderId), zap.Error(err)) + logger.Logger.Error("[MintOrder] Step 3.0 失败: 获取订单错误", + zap.String("order_id", req.OrderId), zap.Error(err)) return fmt.Errorf("order not found: %w", err) } - logger.Logger.Info("[MintOrder] Step 3.0 查询成功", zap.String("order_id", existing.OrderID), zap.String("status", existing.Status)) + logger.Logger.Info("[MintOrder] Step 3.0 查询成功", + zap.String("order_id", existing.OrderID), + zap.String("status", existing.Status)) if existing.Status != models.MintOrderStatusPending { return fmt.Errorf("订单状态为%s，不能继续铸造", existing.Status) } mintOrder = existing - logger.Logger.Info("[MintOrder] Step 3.0 完成: 订单状态", zap.String("status", existing.Status)) // 若阶段二传了字段，则覆盖阶段一的存储值（允许再次编辑元数据） if req.MaterialUrl != "" { @@ -313,7 +342,6 @@ func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, st if getStringValue(mintOrder.MaterialURL) == "" { return fmt.Errorf("material_url is required") } - // info 为必填 if getStringValue(mintOrder.Info) == "" { return fmt.Errorf("info is required") } @@ -324,38 +352,18 @@ func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, st } // 3.1 获取铸造消耗配置（阶梯计价） - // 本次铸造是第 currentMintCount+1 次 - logger.Logger.Info("[MintOrder] Step 3.1: 获取铸造配置", zap.Int32("current_mint_count", currentMintCount)) - var localMintCost *models.MintCostConfig localMintCost, err = s.GetMintCost(currentMintCount + 1) if err != nil { logger.Logger.Error("[MintOrder] Step 3.1 失败", zap.Error(err)) return fmt.Errorf("获取铸造消耗配置失败: %w", err) } - capturedCostCrystal = localMintCost.CostCrystal // 捕获铸造消耗 - logger.Logger.Info("[MintOrder] Step 3.1 完成", zap.Int64("cost_crystal", localMintCost.CostCrystal)) + costCrystal = localMintCost.CostCrystal - // 3.2 扣除水晶余额（调用 User Service RPC） - logger.Logger.Info("[MintOrder] Step 3.2: 扣除水晶", zap.Int64("user_id", userID), zap.Int64("star_id", starID), zap.Int64("delta", -localMintCost.CostCrystal)) - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - capturedBalance, err = s.userClient.UpdateCrystalBalance(ctx, userID, starID, -localMintCost.CostCrystal, - "mint_cost", req.OrderId, fmt.Sprintf("铸造藏品 #%s", req.OrderId)) - if err != nil { - logger.Logger.Error("[MintOrder] Step 3.2 失败: 扣除水晶错误", zap.Error(err)) - return fmt.Errorf("水晶余额不足或扣除失败: %w", err) - } - - logger.Logger.Info("[MintOrder] Step 3.2 完成", zap.Int64("new_balance", newBalance)) - - // 3.3 检查是否触发保底（概率触发） - var boostBps int32 = 0 + // 3.3 检查是否触发保底(纯本地随机 + DB 写,事务内合法;见 mint-task-4) + boostBps = 0 if localMintCost.Probability > 0 && localMintCost.RewardValue > 0 { - // ★ 批次1.5: 用 crypto/rand 替换原 time.Now().UnixNano()%100, - // 消除并发同纳秒相同结果与脚本卡点操纵风险。 if rollGuarantee(localMintCost.Probability) { - boostBps = int32(localMintCost.RewardValue) // reward_value 单位是 bps + boostBps = int32(localMintCost.RewardValue) logger.Logger.Info("Mint guarantee triggered", zap.Int64("user_id", userID), zap.Int64("star_id", starID), @@ -365,36 +373,99 @@ func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, st } } - // 3.4 更新用户铸爱次数和收益提升 - logger.Logger.Info("[MintOrder] Step 3.4: 更新铸爱次数", zap.Int64("user_id", userID), zap.Int64("star_id", starID)) - if err := s.UpdateMintCountAndBoost(ctx, tx, userID, starID, boostBps); err != nil { - logger.Logger.Error("[MintOrder] Step 3.4 失败: 更新铸爱次数错误，将回滚事务", zap.Int64("user_id", userID), zap.Int64("star_id", starID), zap.Error(err)) - return fmt.Errorf("failed to update mint count: %w", err) // 回滚事务，不继续 + // 3.4 更新用户铸爱次数和收益提升(纯 DB,无 RPC) + if err := s.UpdateMintCountAndBoost(context.Background(), tx, userID, starID, boostBps); err != nil { + logger.Logger.Error("[MintOrder] Step 3.4 失败", + zap.Int64("user_id", userID), + zap.Int64("star_id", starID), + zap.Error(err)) + return fmt.Errorf("failed to update mint count: %w", err) } - // 3.5 创建资产记录（状态：Active，CoverURL 直接使用 MaterialURL） + // 序列同步(后续会写 asset,提前对齐序列) + if err := syncAssetsIDSequence(tx); err != nil { + logger.Logger.Warn("[MintOrder] sync assets_id_seq failed", zap.Error(err)) + return fmt.Errorf("failed to sync asset id sequence: %w", err) + } + + // ★ 推进订单 PENDING → PROCESSING(写 cost_crystal 快照) + if err := tx.Model(&models.MintOrder{}). + Where("order_id = ?", req.OrderId). + Updates(map[string]interface{}{ + "status": models.MintOrderStatusProcessing, + "cost_crystal": costCrystal, + "updated_at": time.Now().UnixMilli(), + }).Error; err != nil { + return fmt.Errorf("failed to mark PROCESSING: %w", err) + } + mintOrder.Status = models.MintOrderStatusProcessing + mintOrder.CostCrystal = costCrystal + + logger.Logger.Info("[MintOrder] Phase 1 done, status=PROCESSING", + zap.String("order_id", mintOrder.OrderID), + zap.Int64("cost_crystal", costCrystal)) + + return nil + }) + + if err != nil { + // 阶段1 失败:订单本就是 PENDING,自动回滚即正确状态。 + // 用户可重新 PreCreateMintOrder + CreateMintOrder 重试。 + logger.Logger.Error("[MintOrder] Phase 1 failed, order stays PENDING", + zap.String("order_id", req.OrderId), zap.Error(err)) + return nil, fmt.Errorf("phase1 prepare: %w", err) + } + + // ==================================================================== + // 阶段2 (事务外 RPC):扣水晶 + // - 依赖 mint-task-1 的 source_id 幂等保护,重试安全。 + // - 失败 → markMintOrderFailed(独立事务,标 FAILED,不回调水晶)。 + // - 已扣 → txn_local#2 继续。 + // ==================================================================== + var newBalance int64 + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + logger.Logger.Info("[MintOrder] Step 3.2: 扣除水晶(事务外)", + zap.Int64("user_id", userID), + zap.Int64("star_id", starID), + zap.Int64("delta", -costCrystal)) + newBalance, err = s.userClient.UpdateCrystalBalance(ctx, userID, starID, -costCrystal, + "mint_cost", req.OrderId, fmt.Sprintf("铸造藏品 #%s", req.OrderId)) + if err != nil { + logger.Logger.Error("[MintOrder] Step 3.2 失败: 扣除水晶错误", + zap.String("order_id", req.OrderId), + zap.Error(err)) + // ★ 阶段2 失败:独立事务标 FAILED,不让订单卡在 PROCESSING 永远等 + s.markMintOrderFailed(req.OrderId, fmt.Errorf("phase2 rpc: %w", err)) + return nil, fmt.Errorf("水晶扣费失败: %w", err) + } + logger.Logger.Info("[MintOrder] Step 3.2 完成", + zap.Int64("new_balance", newBalance)) + + // ==================================================================== + // 阶段3 (txn_local#2):建 asset + registry + mint_order → SUCCESS + // - 失败 → markMintOrderFailed(水晶已扣,走对账任务;不在本路径退款) + // ==================================================================== + var asset *models.Asset + err = s.db.Transaction(func(tx *gorm.DB) error { + // 3.5 创建资产记录 logger.Logger.Info("[MintOrder] Step 3.5: 创建资产") materialURLValue := getStringValue(mintOrder.MaterialURL) mintedAt := time.Now().UnixMilli() - // ★ 批次1.5: 不再写入伪造的 tx_hash / block_number。 - // 当前铸造不上链，写模拟 hash 会被前端误当作真实链上凭证；tx_hash/block_number 保持 NULL， - // 待真正上链时再由链上回调回填。mintedAt 表示业务生效时间（非链上时间），保留。 - logger.Logger.Info("[MintOrder] 创建资产参数", zap.Int64("user_id", userID), zap.Int64("star_id", starID), zap.String("name", getStringValue(mintOrder.Name)), zap.String("material_url", materialURLValue)) asset = &models.Asset{ OwnerUID: userID, StarID: starID, Name: getStringValue(mintOrder.Name), - CoverURL: materialURLValue, // 直接使用素材图作为封面 - MaterialURL: &materialURLValue, // 使用指针 + CoverURL: materialURLValue, + MaterialURL: &materialURLValue, Description: mintOrder.Description, Visibility: models.AssetVisibilityPrivate, - Status: models.AssetStatusActive, // 直接设为 Active + Status: models.AssetStatusActive, LikeCount: 0, Info: getStringValue(mintOrder.Info), MintedAt: &mintedAt, } - - // 可选字段 if req.Grade != 0 { asset.Grade = int32ToPtr(req.Grade) } @@ -402,21 +473,15 @@ func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, st asset.Tags = models.StringArray(req.Tags) } - if err := syncAssetsIDSequence(tx); err != nil { - logger.Logger.Warn("[MintOrder] sync assets_id_seq failed", zap.Error(err)) - return fmt.Errorf("failed to sync asset id sequence: %w", err) - } - - logger.Logger.Info("[MintOrder] 执行 tx.Create(asset)") if err := tx.Create(asset).Error; err != nil { - logger.Logger.Error("[MintOrder] Step 3.5 失败: 创建资产错误", zap.String("material_url", materialURLValue), zap.Error(err)) + logger.Logger.Error("[MintOrder] Step 3.5 失败: 创建资产错误", + zap.String("material_url", materialURLValue), zap.Error(err)) return fmt.Errorf("failed to create asset: %w", err) } logger.Logger.Info("Asset created", - zap.Int64("asset_id", asset.ID), - zap.Int64("user_id", userID), - ) - grade := int32(1) // 普通藏品初始等级为1 + zap.Int64("asset_id", asset.ID), zap.Int64("user_id", userID)) + + grade := int32(1) registry := &models.AssetRegistry{ AssetID: asset.ID, AssetType: models.AssetTypeRegular, @@ -432,22 +497,19 @@ func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, st logger.Logger.Error("Failed to create asset registry", zap.Int64("asset_id", asset.ID), zap.Int64("user_id", userID), - zap.Error(err), - ) + zap.Error(err)) return fmt.Errorf("failed to create asset registry: %w", err) } - logger.Logger.Info("Asset registry created", zap.Int64("asset_id", asset.ID), zap.Int64("user_id", userID), - zap.Int32("grade", grade), - ) + zap.Int32("grade", grade)) - // 3.5 推进阶段一订单到 SUCCESS，并关联资产（铸造同步完成） + // 推进订单 PROCESSING → SUCCESS,关联资产 mintedAt = time.Now().UnixMilli() updates := map[string]interface{}{ "asset_id": asset.ID, - "status": models.MintOrderStatusSuccess, // 直接设为成功，无需异步处理 + "status": models.MintOrderStatusSuccess, "cost_crystal": localMintCost.CostCrystal, "error_message": nil, "material_url": getStringValue(mintOrder.MaterialURL), @@ -456,7 +518,7 @@ func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, st "material_type": getStringValue(mintOrder.MaterialType), "event": getStringValue(mintOrder.Event), "info": getStringValue(mintOrder.Info), - "minted_at": mintedAt, // 同步设置上链时间 + "minted_at": mintedAt, } if err := tx.Model(&models.MintOrder{}). Where("order_id = ? AND user_id = ? AND star_id = ?", mintOrder.OrderID, userID, starID). @@ -470,14 +532,16 @@ func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, st logger.Logger.Info("Mint order created", zap.String("order_id", mintOrder.OrderID), - zap.Int64("asset_id", asset.ID), - ) - + zap.Int64("asset_id", asset.ID)) return nil }) if err != nil { - return nil, err + // ★ 阶段3 失败:标 FAILED,水晶已扣由 ReconcileStuckMintOrders 补单 + logger.Logger.Error("[MintOrder] Phase 3 failed, marking FAILED", + zap.String("order_id", req.OrderId), zap.Error(err)) + s.markMintOrderFailed(req.OrderId, fmt.Errorf("phase3 finalize: %w", err)) + return nil, fmt.Errorf("phase2 finalize: %w", err) } // 4. 初始化资产等级记录 @@ -489,9 +553,7 @@ func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, st } } - // 5. 无需异步 AI 处理，cover_url 已在步骤 3.2 中直接设置 - - // 5. 获取所有者的昵称和头像（创建时所有者就是当前用户） + // 5. 获取所有者的昵称和头像(创建时所有者就是当前用户) var ownerNickname string var ownerAvatar string profile, err := s.userClient.GetFanProfile(context.Background(), userID, starID) @@ -508,7 +570,6 @@ func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, st } // 6. 构建响应 - // 获取本次铸造消耗（从事务内捕获的值） response := &pb.CreateMintOrderResponse{ Base: &pbCommon.BaseResponse{ Code: uint32(codes.OK), @@ -516,8 +577,8 @@ func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, st Timestamp: time.Now().UnixMilli(), }, Order: ModelToProtoMintOrder(mintOrder), - Asset: ModelToProtoAssetDetail(asset, ownerNickname, ownerAvatar, false, 0, 0, 0, 0, getInt32Value(asset.Grade)), // 新创建的资产，is_liked 为 false，display_status 默认为 0，earnings、hourlyEarnings 和 exhibitionExpireAt 为 0，grade 从 asset.Grade 获取 - CostCrystal: capturedCostCrystal, + Asset: ModelToProtoAssetDetail(asset, ownerNickname, ownerAvatar, false, 0, 0, 0, 0, getInt32Value(asset.Grade)), + CostCrystal: costCrystal, BalanceAfter: newBalance, } @@ -525,11 +586,11 @@ func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, st zap.String("order_id", mintOrder.OrderID), zap.Int64("asset_id", asset.ID), zap.Int64("user_id", userID), - zap.Int64("cost_crystal", capturedCostCrystal), + zap.Int64("cost_crystal", costCrystal), zap.Int64("balance_after", newBalance), ) - // 事件埋点：asset.mint（fire-and-forget） + // 事件埋点:asset.mint(fire-and-forget) statistic.Get().TrackEvent(context.Background(), &eventPb.Event{ EventType: "asset.mint", UserId: userID, @@ -544,6 +605,37 @@ func (s *mintService) CreateMintOrder(req *pb.CreateMintOrderRequest, userID, st return response, nil } +// markMintOrderFailed 独立事务中标 FAILED + 写入 error_message。 +// +// 失败仅日志,不阻塞主流程(因为 markMintOrderFailed 自身失败时, +// ReconcileStuckMintOrders 也会兜底)。 +// +// ★ Minor#6 nil 防御:cause == nil 时不能 cause.Error() panic, +// 落 "unknown error" 占位,避免把订单标 FAILED 但 error_message 为 NULL。 +func (s *mintService) markMintOrderFailed(orderID string, cause error) { + if orderID == "" { + return + } + if cause == nil { + cause = errors.New("unknown error") + } + now := time.Now().UnixMilli() + msg := cause.Error() + err := s.db.Model(&models.MintOrder{}). + Where("order_id = ?", orderID). + Updates(map[string]interface{}{ + "status": models.MintOrderStatusFailed, + "error_message": msg, + "updated_at": now, + }).Error + if err != nil { + logger.Logger.Error("markMintOrderFailed failed", + zap.String("order_id", orderID), + zap.Error(err), + ) + } +} + // ModelToProtoMintOrder 将数据库模型转换为Proto格式 func ModelToProtoMintOrder(order *models.MintOrder) *pb.MintOrder { if order == nil { @@ -1043,3 +1135,330 @@ func rollGuarantee(probability int64) bool { } return v.Int64() < probability } + +// ===================================================================== +// 孤儿订单对账 (mint-task-3) +// +// 背景: +// CreateMintOrder 拆为三段后,如果资产服务在阶段2(RPC 扣费)和阶段3(建 asset) +// 之间崩溃,可能留下"水晶已扣但 asset 未建"的订单卡在 PROCESSING 状态。 +// +// 如果 markMintOrderFailed 也失败(罕见的 DB 故障),同样会留下 PROCESSING 订单。 +// +// 设计: +// 1) 扫描 status=PROCESSING 且 updated_at 早于 staleAfter 的订单; +// 2) 对每单,查 crystal_transaction_records 是否有该 order_id 的 mint_cost 流水; +// 3) 决策: +// - 已扣水晶 + 无 asset(资产已建但订单状态没推进) → 幂等补完 phase3; +// - 未扣水晶 → 安全标 FAILED(没扣钱,直接结束状态机); +// - 已扣 + 已建 asset(理论上不会发生) → 不动,留人工排查。 +// +// 边界: +// - 幂等:重复跑不会重复建 asset,靠 mint_order.asset_id / asset_registry.asset_id 判断。 +// - 不退款:避免双花。只标 FAILED 或补完 SUCCESS,绝不调用 UpdateCrystalBalance 正向补扣。 +// +// 调用方: +// - 可由 main.go 启动时起一个 goroutine 周期触发(建议 5~15 分钟一次); +// - 也提供 RPC / HTTP 管理接口供运维手动触发(本任务未实现)。 +// ===================================================================== + +// reconcileThreshold 对账扫描阈值(默认 10 分钟前还在 PROCESSING 的订单视为孤儿) +// 与 brief 推荐一致。函数签名上仍允许调用方传入 staleAfter 覆盖默认。 +const reconcileThreshold = 10 * time.Minute + +// ReconcileStuckMintOrders 扫描 status=PROCESSING 且 updated_at 早于 staleAfter 的订单, +// 对每单判定: +// - 水晶已扣(mint_cost 流水存在) + 无 asset → 幂等补完 phase3(建 asset + 订单→SUCCESS) +// - 水晶未扣 → 标 FAILED(没扣钱,安全结束状态机) +// - 水晶已扣 + asset 已建(理论上不会发生) → 不动,告警日志留人工排查 +// +// 幂等:重复跑不会重复建 asset,不会二次扣水晶,不会把 SUCCESS/FAILED 订单再次处理。 +// +// 返回:扫描到的 PROCESSING 订单数,以及最终处理结果(recovered / marked_failed / skipped)。 +func (s *mintService) ReconcileStuckMintOrders(ctx context.Context, staleAfter time.Duration) error { + if staleAfter <= 0 { + staleAfter = reconcileThreshold + } + cutoff := time.Now().Add(-staleAfter).UnixMilli() + + // 1) 扫描陈旧 PROCESSING 订单(每次最多处理 100 单,避免长事务) + var stuckOrders []models.MintOrder + if err := s.db.WithContext(ctx). + Where("status = ? AND updated_at < ?", models.MintOrderStatusProcessing, cutoff). + Order("updated_at ASC"). + Limit(100). + Find(&stuckOrders).Error; err != nil { + return fmt.Errorf("scan stuck mint orders: %w", err) + } + + logger.Logger.Info("ReconcileStuckMintOrders start", + zap.Int("stuck_count", len(stuckOrders)), + zap.Int64("cutoff_ms", cutoff), + zap.Duration("stale_after", staleAfter)) + + var recovered, failed, skipped int + for i := range stuckOrders { + order := stuckOrders[i] + action, err := s.reconcileOneMintOrder(ctx, &order) + if err != nil { + // 单单失败不中断对账(其他订单还有救) + failed++ + logger.Logger.Error("reconcile one order failed", + zap.String("order_id", order.OrderID), + zap.Error(err)) + continue + } + switch action { + case reconcileActionRecovered: + recovered++ + case reconcileActionFailed: + failed++ + default: + skipped++ + } + } + + logger.Logger.Info("ReconcileStuckMintOrders done", + zap.Int("scanned", len(stuckOrders)), + zap.Int("recovered", recovered), + zap.Int("marked_failed", failed), + zap.Int("skipped", skipped)) + + return nil +} + +// reconcileAction 对账动作结果(给 worker 上报计数用) +type reconcileAction int + +const ( + reconcileActionSkipped reconcileAction = iota + reconcileActionFailed // 安全标 FAILED(无 mint_cost 流水) + reconcileActionRecovered // 幂等补完 SUCCESS +) + +// reconcileOneMintOrder 处理单个孤儿订单。 +// +// 决策表: +// +// | asset_id | mint_cost 流水 | 动作 | +// | -------- | -------------- | -------------------------- | +// | NULL | 无 | markMintOrderFailed (FAILED) | +// | NULL | 有 | 补完 phase3 (Recovered) | +// | 已设 | (任意) | force SUCCESS (Recovered) | +// +// 返回 action + error:error 不为 nil 时 action 忽略不计(failed)。 +func (s *mintService) reconcileOneMintOrder(ctx context.Context, order *models.MintOrder) (reconcileAction, error) { + // 二次校验:防止扫描后被并发改状态 + fresh, err := s.mintOrderRepo.GetByOrderID(order.OrderID) + if err != nil { + return reconcileActionSkipped, fmt.Errorf("reload order: %w", err) + } + if fresh.Status != models.MintOrderStatusProcessing { + logger.Logger.Debug("reconcile: order no longer PROCESSING, skip", + zap.String("order_id", order.OrderID), + zap.String("status", fresh.Status)) + return reconcileActionSkipped, nil + } + + // 查水晶流水:source_id = order_id, change_type = 'mint_cost' + var txCount int64 + if err := s.db.WithContext(ctx). + Table("crystal_transaction_records"). + Where("source_id = ? AND change_type = ?", order.OrderID, "mint_cost"). + Count(&txCount).Error; err != nil { + return reconcileActionSkipped, fmt.Errorf("count crystal tx: %w", err) + } + + // 决策分支 + if txCount == 0 { + // 未扣水晶:安全标 FAILED + logger.Logger.Info("reconcile: no crystal charged, mark FAILED", + zap.String("order_id", order.OrderID), + zap.Int64("user_id", order.UserID), + zap.Int64("star_id", order.StarID)) + s.markMintOrderFailed(order.OrderID, fmt.Errorf("reconciled: PROCESSING without mint_cost flow")) + return reconcileActionFailed, nil + } + + // 已扣水晶 + if fresh.AssetID != nil && *fresh.AssetID > 0 { + // 理论上不该发生:asset 已建但订单 PROCESSING。 + // 直接推进到 SUCCESS 兜底,避免无限对账。 + logger.Logger.Warn("reconcile: order has asset_id but status=PROCESSING, force SUCCESS", + zap.String("order_id", order.OrderID), + zap.Int64("asset_id", *fresh.AssetID)) + now := time.Now().UnixMilli() + if err := s.db.Model(&models.MintOrder{}). + Where("order_id = ? AND status = ?", order.OrderID, models.MintOrderStatusProcessing). + Updates(map[string]interface{}{ + "status": models.MintOrderStatusSuccess, + "updated_at": now, + }).Error; err != nil { + return reconcileActionSkipped, fmt.Errorf("force SUCCESS: %w", err) + } + return reconcileActionRecovered, nil + } + + // 已扣水晶 + 无 asset:幂等补完 phase3(带行锁) + logger.Logger.Info("reconcile: crystal charged but no asset, recovering phase3", + zap.String("order_id", order.OrderID), + zap.Int64("user_id", order.UserID), + zap.Int64("star_id", order.StarID)) + + if err := s.finalizeMintOrderRecovery(ctx, order.OrderID); err != nil { + return reconcileActionSkipped, err + } + return reconcileActionRecovered, nil +} + +// finalizeMintOrderRecovery 在对账场景下补完 phase3(建 asset + registry + 订单→SUCCESS)。 +// +// 与 CreateMintOrder 阶段3 的区别: +// - 不再调 UpdateCrystalBalance(钱已扣过) +// - 不再 UpdateMintCountAndBoost(阶段1 已更新过) +// - 复用同样的事务结构 + 序列同步 +// - ★ mint-task-3 review: 事务**开头**对该 order 加 SELECT...FOR UPDATE 行锁, +// 锁内重新判定 status/asset_id,避免与并发的 CreateMintOrder-phase3 双建 asset。 +// +// 失败处理:写 ERROR 日志。下一轮对账或人工介入, +// 关键原则:**绝不退款,避免双花**。 +func (s *mintService) finalizeMintOrderRecovery(ctx context.Context, orderID string) error { + now := time.Now().UnixMilli() + var createdAssetID int64 + var recoveredUserID, recoveredStarID int64 + + err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + // ★ 行锁:SELECT...FOR UPDATE 串行化并发的 finalize / CreateMintOrder-phase3 + var locked models.MintOrder + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). + Where("order_id = ?", orderID). + First(&locked).Error; err != nil { + return fmt.Errorf("lock order for recovery: %w", err) + } + + // 锁内重判:状态机是否已被并发推进? + if locked.Status != models.MintOrderStatusProcessing { + logger.Logger.Info("recovery: order no longer PROCESSING after lock, skip", + zap.String("order_id", orderID), + zap.String("status", locked.Status)) + return nil + } + + recoveredUserID = locked.UserID + recoveredStarID = locked.StarID + + // 锁内重判:asset 已被并发路径建好了? + hasAsset := false + var existingAsset models.Asset + if locked.AssetID != nil && *locked.AssetID > 0 { + if terr := tx.Where("id = ?", *locked.AssetID).First(&existingAsset).Error; terr == nil { + hasAsset = true + createdAssetID = existingAsset.ID + } + } + + if !hasAsset { + // 序列同步 + if serr := syncAssetsIDSequence(tx); serr != nil { + return fmt.Errorf("sync asset id sequence: %w", serr) + } + + // 重建 asset(与正常 CreateMintOrder 阶段3 一致) + materialURLValue := getStringValue(locked.MaterialURL) + nameValue := getStringValue(locked.Name) + if nameValue == "" { + nameValue = "未命名藏品" + } + infoValue := getStringValue(locked.Info) + mintedAt := now + asset := &models.Asset{ + OwnerUID: locked.UserID, + StarID: locked.StarID, + Name: nameValue, + CoverURL: materialURLValue, + MaterialURL: &materialURLValue, + Description: locked.Description, + Visibility: models.AssetVisibilityPrivate, + Status: models.AssetStatusActive, + LikeCount: 0, + Info: infoValue, + MintedAt: &mintedAt, + } + if err := tx.Create(asset).Error; err != nil { + return fmt.Errorf("create asset during recovery: %w", err) + } + createdAssetID = asset.ID + + grade := int32(1) + registry := &models.AssetRegistry{ + AssetID: asset.ID, + AssetType: models.AssetTypeRegular, + OwnerUID: locked.UserID, + StarID: locked.StarID, + Grade: &grade, + Status: models.AssetRegistryStatusActive, + LikeCount: 0, + CreatedAt: now, + UpdatedAt: now, + } + if err := tx.Create(registry).Error; err != nil { + return fmt.Errorf("create asset registry during recovery: %w", err) + } + } + + // 推进订单 PROCESSING → SUCCESS(带 status 守卫,防双写) + updates := map[string]interface{}{ + "status": models.MintOrderStatusSuccess, + "updated_at": now, + } + if createdAssetID > 0 { + updates["asset_id"] = createdAssetID + updates["minted_at"] = now + } + // mint_count 在阶段1 已自增过,这里不再回退(防止 mint_count 倒退) + // 同样不重置 cost_crystal(已记录的快照代表真实扣费) + + res := tx.Model(&models.MintOrder{}). + Where("order_id = ? AND status = ?", orderID, models.MintOrderStatusProcessing). + Updates(updates) + if res.Error != nil { + return fmt.Errorf("mark SUCCESS during recovery: %w", res.Error) + } + if res.RowsAffected == 0 { + // 理论上锁内不可能被改;RowsAffected=0 极少见,留 WARN 不算错。 + logger.Logger.Warn("recovery: status guard returned 0 rows, status may have flipped", + zap.String("order_id", orderID)) + return nil + } + + // 初始化资产等级记录(如果有 service) + if s.assetLevelService != nil && createdAssetID > 0 { + if _, lerr := s.assetLevelService.GetOrCreateRecord(createdAssetID); lerr != nil { + // 不阻塞主流程,只 WARN + logger.Logger.Warn("recovery: failed to create asset level record", + zap.Int64("asset_id", createdAssetID), zap.Error(lerr)) + } + } + + return nil + }) + + if err != nil { + logger.Logger.Error("recovery transaction failed, will retry next round", + zap.String("order_id", orderID), + zap.Int64("user_id", recoveredUserID), + zap.Int64("star_id", recoveredStarID), + zap.Error(err)) + // 不标 FAILED — 留到下轮重试或人工排查。 + // 因为水晶已扣,如果本轮标 FAILED,运营只能"手动重建 asset"或"退款", + // 都比"再让对账跑一次"风险更高。 + return err + } + + logger.Logger.Info("recovery succeeded", + zap.String("order_id", orderID), + zap.Int64("asset_id", createdAssetID), + zap.Int64("user_id", recoveredUserID), + zap.Int64("star_id", recoveredStarID)) + return nil +} diff --git a/backend/services/assetService/service/mint_service_idempotency_test.go b/backend/services/assetService/service/mint_service_idempotency_test.go index d62f9f8..9709f8a 100644 --- a/backend/services/assetService/service/mint_service_idempotency_test.go +++ b/backend/services/assetService/service/mint_service_idempotency_test.go @@ -3,27 +3,39 @@ package service import ( "context" "crypto/rand" + "fmt" "math/big" "testing" + "time" "github.com/topfans/backend/pkg/models" pb "github.com/topfans/backend/pkg/proto/asset" pbUser "github.com/topfans/backend/pkg/proto/user" "github.com/topfans/backend/services/assetService/client" "github.com/topfans/backend/services/assetService/repository" + "gorm.io/gorm" ) // mockUserClient 服务层测试用的 UserServiceClient mock。 // // 跟踪 UpdateCrystalBalance 调用次数,确保幂等短路路径不会重复扣费。 // GetFanProfile 返回当前 balance(测试中只检查不被二次扣即可)。 +// +// ★ mint-task-3: 新增 failOnUpdate 字段用于模拟"事务外 RPC 失败"路径, +// 验证 CreateMintOrder 在 RPC 失败时仍把 mint_order 推进到 FAILED 状态, +// 而不是回滚成 PENDING 留下僵尸订单。 type mockUserClient struct { balance int64 updateCrystalCalls int + failOnUpdate bool } func (m *mockUserClient) UpdateCrystalBalance(_ context.Context, _, _ int64, delta int64, _, _, _ string) (int64, error) { m.updateCrystalCalls++ + if m.failOnUpdate { + // 模拟 userService 侧的真实失败(余额不足/服务降级/超时等) + return m.balance, fmt.Errorf("simulated RPC failure") + } m.balance += delta return m.balance, nil } @@ -236,4 +248,253 @@ func TestRollGuarantee_Distribution(t *testing.T) { t.Errorf("rollGuarantee(50) over %d trials: hits=%d, want in [%d,%d] (≈50%%±9σ)", n, hits, minHits, maxHits) } -} \ No newline at end of file +} + +// TestCreateMintOrder_RPCFailure_MarksOrderFailed 验证 ★ mint-task-3 重构: +// +// 当事务外 RPC (UpdateCrystalBalance) 失败时,mint_order 必须落到 FAILED 状态 +// 并写入 error_message,而不是回滚成 PENDING 形成"僵尸订单"。 +// +// 旧实现: RPC 在事务内,失败 → 事务回滚 → 订单仍是 PENDING (水/状态不对齐, +// 用户看到订单还在 PENDING,但实际钱没扣,容易被重放造成状态机错位)。 +// 新实现: 阶段1 PENDING→PROCESSING; 阶段2 RPC 失败 → 独立事务标 FAILED; +// 阶段3 不再走(订单已 FAILED,用户必须 PreCreateMintOrder 新建)。 +// +// 期望: +// - CreateMintOrder 返回 error +// - mint_order.Status = FAILED +// - mint_order.error_message 非空(记录失败原因) +// - mint_orders 表里残留 PROCESSING 状态订单 = 0 +func TestCreateMintOrder_RPCFailure_MarksOrderFailed(t *testing.T) { + db := setupServiceTestDB(t) + defer cleanupServiceTestDB(t, db) + + star := createServiceTestStar(t, db, "mint_rpc_fail_star") + user := createServiceTestUser(t, db, "19900077002") + + // ★ RPC 失败路径:failOnUpdate=true 模拟 userService 报余额不足 / 超时 + uc := &mockUserClient{balance: 100, failOnUpdate: true} + // localMintCostRepo / userMintCountRepo 用真 repo 走 DB(都只做只读查询,不会脏写) + svc := NewMintService( + repository.NewAssetRepository(db), + repository.NewMintOrderRepository(db), + uc, + db, nil, + nil, + repository.NewMintCostRepository(), + repository.NewUserMintCountRepository(), + nil, + ) + + // 阶段一:预创建 PENDING 订单(由调用方先调 PreCreateMintOrder 拿 order_id) + const orderID = "rpc-fail-uuid-001" + if err := db.Create(&models.MintOrder{ + OrderID: orderID, + UserID: user.ID, + StarID: star.StarID, + Status: models.MintOrderStatusPending, + }).Error; err != nil { + t.Fatalf("seed mint_order PENDING: %v", err) + } + + // 阶段二:CreateMintOrder 应该因 RPC 失败返回 error,并把订单落 FAILED + _, err := svc.CreateMintOrder(pbCreateMintReq(orderID, user.ID, star.StarID), user.ID, star.StarID) + if err == nil { + t.Fatal("expected error from RPC failure, got nil") + } + + // ★ 关键断言:订单状态机必须推进到 FAILED(不是 PENDING 也不是 PROCESSING) + finalOrder, ferr := repository.NewMintOrderRepository(db).GetByOrderID(orderID) + if ferr != nil { + t.Fatalf("failed to reload mint_order: %v", ferr) + } + if finalOrder.Status != models.MintOrderStatusFailed { + t.Errorf("want status=FAILED after RPC failure, got %s", finalOrder.Status) + } + if finalOrder.ErrorMessage == nil || *finalOrder.ErrorMessage == "" { + t.Error("want error_message populated on FAILED order, got nil/empty") + } + + // ★ 关键断言:不能有残留的 PROCESSING 僵尸订单(同一 order_id) + var stuckCount int64 + if err := db.Model(&models.MintOrder{}). + Where("order_id = ? AND status = ?", orderID, models.MintOrderStatusProcessing). + Count(&stuckCount).Error; err != nil { + t.Fatalf("count stuck: %v", err) + } + if stuckCount != 0 { + t.Errorf("want 0 stuck PROCESSING orders, got %d", stuckCount) + } + + // ★ 幂等补充:已 FAILED 的订单不应被"二次 RPC 扣费" + // (FAILED 状态会从入口拒绝二次 CreateMintOrder,但这里只验证当前流程不重复扣) + if uc.updateCrystalCalls != 1 { + t.Errorf("want exactly 1 UpdateCrystalBalance call (the failing one), got %d", uc.updateCrystalCalls) + } +} + +// TestReconcileStuckMintOrders_NoCrystalCharged_MarksFailed 验证对账场景 1: +// +// PROCESSING 超时订单 + crystal_transaction_records 中无 mint_cost 流水 +// → 安全标 FAILED (钱没扣,直接结束状态机) +// +// 适用于: 阶段2 RPC 调用失败但 markMintOrderFailed 也失败的极端边缘情况, +// 或者是迁移/历史遗留的 PROCESSING 订单。 +func TestReconcileStuckMintOrders_NoCrystalCharged_MarksFailed(t *testing.T) { + db := setupServiceTestDB(t) + defer cleanupServiceTestDB(t, db) + + star := createServiceTestStar(t, db, "reconcile_dry_star") + user := createServiceTestUser(t, db, "19900077003") + + // 造一个"陈旧" PROCESSING 订单:updated_at 调到阈值之前 + // 注意:MintOrder.BeforeCreate hook 会强制把 UpdatedAt 设为 now, + // 必须 Create 后再 Update 一次 raw。 + const orderID = "reconcile-dry-uuid-001" + if err := db.Create(&models.MintOrder{ + OrderID: orderID, + UserID: user.ID, + StarID: star.StarID, + Status: models.MintOrderStatusProcessing, + CostCrystal: 50, + }).Error; err != nil { + t.Fatalf("seed stuck order: %v", err) + } + staleTime := time.Now().Add(-20 * time.Minute).UnixMilli() + if err := db.Model(&models.MintOrder{}). + Where("order_id = ?", orderID). + Update("updated_at", staleTime).Error; err != nil { + t.Fatalf("backdate updated_at: %v", err) + } + + svc := &mintService{ + db: db, + mintOrderRepo: repository.NewMintOrderRepository(db), + } + + // 阈值 10 分钟,订单已 20 分钟未更新,应该被对账命中 + if err := svc.ReconcileStuckMintOrders(context.Background(), 10*time.Minute); err != nil { + t.Fatalf("ReconcileStuckMintOrders failed: %v", err) + } + + final, err := repository.NewMintOrderRepository(db).GetByOrderID(orderID) + if err != nil { + t.Fatalf("reload: %v", err) + } + if final.Status != models.MintOrderStatusFailed { + t.Errorf("want status=FAILED, got %s", final.Status) + } + if final.ErrorMessage == nil || *final.ErrorMessage == "" { + t.Error("want error_message set on reconciled order") + } +} + +// TestReconcileStuckMintOrders_CrystalChargedButNoAsset_Recovers 验证对账场景 2: +// +// PROCESSING 超时订单 + crystal_transaction_records 中有 mint_cost 流水 +// → 幂等补完 phase3 (建 asset + registry + 订单→SUCCESS) +// +// 这是对账最关键的"反向补单"场景: 用户钱已扣,但 asset 因崩溃未建。 +// 必须靠 asset_registry 或 mint_order.asset_id 判断是否已建,绝不二次重建。 +func TestReconcileStuckMintOrders_CrystalChargedButNoAsset_Recovers(t *testing.T) { + db := setupServiceTestDB(t) + defer cleanupServiceTestDB(t, db) + + star := createServiceTestStar(t, db, "reconcile_recover_star") + user := createServiceTestUser(t, db, "19900077004") + + // 造一个"陈旧" PROCESSING 订单:有 cost_crystal,有 material_url/name/info + // 等(说明阶段一已经把订单填全了) + const orderID = "reconcile-recover-uuid-001" + matURL := "http://x/r.jpg" + name := "test-asset" + desc := "test-desc" + info := "test-info" + if err := db.Create(&models.MintOrder{ + OrderID: orderID, + UserID: user.ID, + StarID: star.StarID, + Status: models.MintOrderStatusProcessing, + CostCrystal: 50, + MaterialURL: &matURL, + Name: &name, + Description: &desc, + Info: &info, + MaterialType: stringPtr("new"), + }).Error; err != nil { + t.Fatalf("seed stuck order: %v", err) + } + // backdate updated_at(BeforeCreate hook 会覆写) + staleTime := time.Now().Add(-20 * time.Minute).UnixMilli() + if err := db.Model(&models.MintOrder{}). + Where("order_id = ?", orderID). + Update("updated_at", staleTime).Error; err != nil { + t.Fatalf("backdate updated_at: %v", err) + } + + // 模拟"水晶已扣"的流水记录(source_id = order_id, change_type = mint_cost) + if err := db.Exec(` + INSERT INTO crystal_transaction_records + (user_id, star_id, change_type, delta, balance_before, balance_after, source_id, description, created_at) + VALUES (?, ?, 'mint_cost', -50, 100, 50, ?, '铸造藏品 #reconcile-recover-uuid-001', ?) + `, user.ID, star.StarID, orderID, time.Now().UnixMilli()).Error; err != nil { + t.Fatalf("seed crystal tx: %v", err) + } + + // ★ 对账不应该再调 RPC(钱已扣过);但 mintService 需要非 nil userClient 占位 + uc := &mockUserClient{balance: 50} + svc := &mintService{ + db: db, + assetRepo: repository.NewAssetRepository(db), + mintOrderRepo: repository.NewMintOrderRepository(db), + userClient: uc, + registryRepo: nil, // 不强依赖 registry,补单流程会 skip + } + + if err := svc.ReconcileStuckMintOrders(context.Background(), 10*time.Minute); err != nil { + t.Fatalf("ReconcileStuckMintOrders failed: %v", err) + } + + final, err := repository.NewMintOrderRepository(db).GetByOrderID(orderID) + if err != nil { + t.Fatalf("reload: %v", err) + } + if final.Status != models.MintOrderStatusSuccess { + t.Errorf("want status=SUCCESS after recovery, got %s", final.Status) + } + if final.AssetID == nil { + t.Error("want asset_id backfilled on recovery, got nil") + } + + // ★ 关键:不该再调 UpdateCrystalBalance(钱已扣过) + if uc.updateCrystalCalls != 0 { + t.Errorf("reconcile must not re-charge crystal, got %d calls", uc.updateCrystalCalls) + } + + // ★ 幂等:再跑一次对账,不应再创建第二个 asset + preAssetCount := countAssetsForUser(t, db, user.ID) + if err := svc.ReconcileStuckMintOrders(context.Background(), 10*time.Minute); err != nil { + t.Fatalf("second reconcile: %v", err) + } + postAssetCount := countAssetsForUser(t, db, user.ID) + if preAssetCount != postAssetCount { + t.Errorf("reconcile must be idempotent (no duplicate assets): pre=%d post=%d", + preAssetCount, postAssetCount) + } + if final.AssetID == nil { + t.Error("final order must keep asset_id") + } +} + +// countAssetsForUser 计数测试用户的资产数(对账幂等性验证用) +func countAssetsForUser(t *testing.T, db *gorm.DB, userID int64) int64 { + t.Helper() + var n int64 + if err := db.Model(&models.Asset{}).Where("owner_uid = ?", userID).Count(&n).Error; err != nil { + t.Fatalf("count assets: %v", err) + } + return n +} + +// stringPtr helper(测试用) +func stringPtr(s string) *string { return &s } \ No newline at end of file diff --git a/backend/services/assetService/service/peripheral_test_helpers_test.go b/backend/services/assetService/service/peripheral_test_helpers_test.go index ee1b90a..4e0c23c 100644 --- a/backend/services/assetService/service/peripheral_test_helpers_test.go +++ b/backend/services/assetService/service/peripheral_test_helpers_test.go @@ -66,8 +66,13 @@ func setupServiceTestDB(t *testing.T) *gorm.DB { // // 沿用 repository 包 cleanupTestDB 的语义:清理 mobile LIKE '199%' 的用户与 // 关联数据,以及 identity_id LIKE 'test_service_%' / 'test_peripheral_%' 的明星。 +// +// ★ mint-task-3:增加 crystal_transaction_records(source_id 唯一索引需要),按 source_id +// 含 'reconcile-' / 'rpc-fail-' / 'idem-' 的 order_id 清理,避免跨测试因 unique constraint +// 失败。 func cleanupServiceTestDB(t *testing.T, db *gorm.DB) { t.Helper() + db.Exec("DELETE FROM crystal_transaction_records WHERE user_id IN (SELECT id FROM users WHERE mobile LIKE '199%') OR source_id LIKE 'rpc-%' OR source_id LIKE 'reconcile-%' OR source_id LIKE 'idem-%'") db.Exec("DELETE FROM asset_registry WHERE owner_uid IN (SELECT id FROM users WHERE mobile LIKE '199%')") db.Exec("DELETE FROM peripheral_info WHERE asset_id IN (SELECT id FROM assets WHERE owner_uid IN (SELECT id FROM users WHERE mobile LIKE '199%'))") db.Exec("DELETE FROM asset_likes WHERE user_id IN (SELECT id FROM users WHERE mobile LIKE '199%') OR asset_id IN (SELECT id FROM assets WHERE owner_uid IN (SELECT id FROM users WHERE mobile LIKE '199%'))")