refactor(assetService): CreateMintOrder 3-phase txn + orphan reconciliation (P0-2)
- 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 <noreply@anthropic.com>
This commit is contained in:
zerosaturation
parent
b7022d2dc1
commit
3407e30395
@ -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 自动迁移数据库表
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 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 }
|
||||
@ -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%'))")
|
||||
|
||||
Loading…
Reference in New Issue
Block a user