fix: 代码质量清理 — 统一C端支付枚举、修复静默错误、注入Worker回调、移除废弃同步代码
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 7m6s

- C端订单支付状态直接使用管理端统一枚举(1/2/3/4),移除冗余映射函数
- 资产查询绑定卡/套餐查询失败时记录Warn日志而非静默忽略
- Worker bootstrap注入停复机回调(流量耗尽停机、套餐激活复机)
- 删除废弃的SIM状态同步服务和任务处理器
- 新增开发环境数据清理脚本(full/soft/table三种模式)
This commit is contained in:
2026-03-28 15:27:59 +08:00
parent e81d3a58ad
commit 65e461eff7
10 changed files with 668 additions and 399 deletions

View File

@@ -3,6 +3,7 @@ package bootstrap
import (
"github.com/break/junhong_cmp_fiber/internal/service/commission_calculation"
"github.com/break/junhong_cmp_fiber/internal/service/commission_stats"
iotCardSvc "github.com/break/junhong_cmp_fiber/internal/service/iot_card"
orderSvc "github.com/break/junhong_cmp_fiber/internal/service/order"
packagepkg "github.com/break/junhong_cmp_fiber/internal/service/package"
pollingSvc "github.com/break/junhong_cmp_fiber/internal/service/polling"
@@ -100,6 +101,11 @@ func initWorkerServices(stores *queue.WorkerStores, deps *WorkerDependencies) *q
deps.Logger,
)
// 创建停复机服务并注入回调:流量耗尽自动停机、套餐激活自动复机
stopResumeService := iotCardSvc.NewStopResumeService(deps.DB, deps.Redis, stores.IotCard, stores.DeviceSimBinding, deps.GatewayClient, deps.Logger)
usageService.SetStopResumeCallback(stopResumeService)
activationService.SetResumeCallback(stopResumeService)
return &queue.WorkerServices{
CommissionCalculation: commissionCalculationService,
CommissionStats: commissionStatsService,

View File

@@ -111,11 +111,7 @@ func (h *ClientOrderHandler) ListOrders(c *fiber.Ctx) error {
}
if req.PaymentStatus != nil {
paymentStatus, ok := clientPaymentStatusToOrderStatus(*req.PaymentStatus)
if !ok {
return errors.New(errors.CodeInvalidParam)
}
query = query.Where("payment_status = ?", paymentStatus)
query = query.Where("payment_status = ?", *req.PaymentStatus)
}
var total int64
@@ -160,7 +156,7 @@ func (h *ClientOrderHandler) ListOrders(c *fiber.Ctx) error {
OrderID: order.ID,
OrderNo: order.OrderNo,
TotalAmount: order.TotalAmount,
PaymentStatus: orderStatusToClientPaymentStatus(order.PaymentStatus),
PaymentStatus: order.PaymentStatus,
CreatedAt: formatClientOrderTime(order.CreatedAt),
PackageNames: packageNames,
})
@@ -221,7 +217,7 @@ func (h *ClientOrderHandler) GetOrderDetail(c *fiber.Ctx) error {
OrderID: order.ID,
OrderNo: order.OrderNo,
TotalAmount: order.TotalAmount,
PaymentStatus: orderStatusToClientPaymentStatus(order.PaymentStatus),
PaymentStatus: order.PaymentStatus,
PaymentMethod: order.PaymentMethod,
CreatedAt: formatClientOrderTime(order.CreatedAt),
PaidAt: formatClientOrderTimePtr(order.PaidAt),
@@ -373,32 +369,6 @@ func (h *ClientOrderHandler) getOrderVirtualNo(ctx context.Context, order *model
}
}
func orderStatusToClientPaymentStatus(status int) int {
switch status {
case model.PaymentStatusPending:
return 0
case model.PaymentStatusPaid:
return 1
case model.PaymentStatusCancelled:
return 2
default:
return status
}
}
func clientPaymentStatusToOrderStatus(status int) (int, bool) {
switch status {
case 0:
return model.PaymentStatusPending, true
case 1:
return model.PaymentStatusPaid, true
case 2:
return model.PaymentStatusCancelled, true
default:
return 0, false
}
}
func formatClientOrderTime(t time.Time) string {
if t.IsZero() {
return ""

View File

@@ -25,7 +25,7 @@ type ClientOrderInfo struct {
OrderID uint `json:"order_id" description:"订单ID"`
OrderNo string `json:"order_no" description:"订单号"`
TotalAmount int64 `json:"total_amount" description:"订单总金额(分)"`
PaymentStatus int `json:"payment_status" description:"支付状态 (0:待支付, 1:已支付, 2:已取消)"`
PaymentStatus int `json:"payment_status" description:"支付状态 (1:待支付, 2:已支付, 3:已取消, 4:已退款)"`
CreatedAt string `json:"created_at" description:"创建时间"`
}
@@ -63,7 +63,7 @@ type LinkedPackageInfo struct {
// ClientOrderListRequest D2 客户端订单列表请求
type ClientOrderListRequest struct {
Identifier string `json:"identifier" query:"identifier" validate:"required,min=1,max=50" required:"true" minLength:"1" maxLength:"50" description:"资产标识符SN/IMEI/虚拟号/ICCID/MSISDN"`
PaymentStatus *int `json:"payment_status" query:"payment_status" validate:"omitempty,min=0,max=2" minimum:"0" maximum:"2" description:"支付状态 (0:待支付, 1:已支付, 2:已取消)"`
PaymentStatus *int `json:"payment_status" query:"payment_status" validate:"omitempty,min=1,max=4" minimum:"1" maximum:"4" description:"支付状态 (1:待支付, 2:已支付, 3:已取消, 4:已退款)"`
Page int `json:"page" query:"page" validate:"required,min=1" required:"true" minimum:"1" description:"页码"`
PageSize int `json:"page_size" query:"page_size" validate:"required,min=1,max=100" required:"true" minimum:"1" maximum:"100" description:"每页数量"`
}
@@ -73,7 +73,7 @@ type ClientOrderListItem struct {
OrderID uint `json:"order_id" description:"订单ID"`
OrderNo string `json:"order_no" description:"订单号"`
TotalAmount int64 `json:"total_amount" description:"订单总金额(分)"`
PaymentStatus int `json:"payment_status" description:"支付状态 (0:待支付, 1:已支付, 2:已取消)"`
PaymentStatus int `json:"payment_status" description:"支付状态 (1:待支付, 2:已支付, 3:已取消, 4:已退款)"`
CreatedAt string `json:"created_at" description:"创建时间"`
PackageNames []string `json:"package_names" description:"套餐名称列表"`
}
@@ -95,7 +95,7 @@ type ClientOrderDetailResponse struct {
OrderID uint `json:"order_id" description:"订单ID"`
OrderNo string `json:"order_no" description:"订单号"`
TotalAmount int64 `json:"total_amount" description:"订单总金额(分)"`
PaymentStatus int `json:"payment_status" description:"支付状态 (0:待支付, 1:已支付, 2:已取消)"`
PaymentStatus int `json:"payment_status" description:"支付状态 (1:待支付, 2:已支付, 3:已取消, 4:已退款)"`
PaymentMethod string `json:"payment_method" description:"支付方式"`
CreatedAt string `json:"created_at" description:"创建时间"`
PaidAt *string `json:"paid_at,omitempty" description:"支付时间"`

View File

@@ -133,7 +133,12 @@ func (s *Service) buildDeviceResolveResponse(ctx context.Context, device *model.
slotMap[b.IotCardID] = b.SlotPosition
isCurrentMap[b.IotCardID] = b.IsCurrent
}
cards, _ := s.iotCardStore.GetByIDs(ctx, cardIDs)
cards, err := s.iotCardStore.GetByIDs(ctx, cardIDs)
if err != nil {
logger.GetAppLogger().Warn("查询设备绑定卡信息失败,结果可能不完整",
zap.Uints("card_ids", cardIDs),
zap.Error(err))
}
for _, c := range cards {
resp.Cards = append(resp.Cards, dto.BoundCardInfo{
CardID: c.ID,
@@ -285,7 +290,12 @@ func (s *Service) GetRealtimeStatus(ctx context.Context, assetType string, id ui
slotMap[b.IotCardID] = b.SlotPosition
isCurrentMap[b.IotCardID] = b.IsCurrent
}
cards, _ := s.iotCardStore.GetByIDs(ctx, cardIDs)
cards, err := s.iotCardStore.GetByIDs(ctx, cardIDs)
if err != nil {
logger.GetAppLogger().Warn("查询设备绑定卡信息失败,结果可能不完整",
zap.Uints("card_ids", cardIDs),
zap.Error(err))
}
for _, c := range cards {
resp.Cards = append(resp.Cards, dto.BoundCardInfo{
CardID: c.ID,
@@ -533,7 +543,12 @@ func (s *Service) GetPackages(ctx context.Context, assetType string, id uint, pa
for id := range pkgIDSet {
pkgIDs = append(pkgIDs, id)
}
packages, _ := s.packageStore.GetByIDsUnscoped(ctx, pkgIDs)
packages, pkgErr := s.packageStore.GetByIDsUnscoped(ctx, pkgIDs)
if pkgErr != nil {
logger.GetAppLogger().Warn("批量查询套餐信息失败,套餐名称可能缺失",
zap.Uints("package_ids", pkgIDs),
zap.Error(pkgErr))
}
pkgMap := make(map[uint]*model.Package, len(packages))
for _, p := range packages {
pkgMap[p.ID] = p

View File

@@ -311,7 +311,7 @@ func (s *Service) createPackageOrder(
OrderID: order.ID,
OrderNo: order.OrderNo,
TotalAmount: order.TotalAmount,
PaymentStatus: orderStatusToClientStatus(order.PaymentStatus),
PaymentStatus: order.PaymentStatus,
CreatedAt: formatClientServiceTime(order.CreatedAt),
},
PayConfig: buildClientPayConfig(appID, payResult.PayConfig),
@@ -641,19 +641,6 @@ func buildClientPurchaseBusinessKey(customerID uint, assetInfo *dto.AssetResolve
return fmt.Sprintf("%d:%s:%d:%s:%s", customerID, assetInfo.AssetType, assetInfo.AssetID, req.AppType, strings.Join(parts, ","))
}
func orderStatusToClientStatus(status int) int {
switch status {
case model.PaymentStatusPending:
return 0
case model.PaymentStatusPaid:
return 1
case model.PaymentStatusCancelled:
return 2
default:
return status
}
}
func rechargeStatusToClientStatus(status int) int {
switch status {
case 1:

View File

@@ -1,170 +0,0 @@
// Package sync 提供数据同步的业务逻辑服务
// 包含批量数据同步、任务调度等功能
package sync
import (
"context"
"fmt"
"github.com/break/junhong_cmp_fiber/internal/task"
"github.com/break/junhong_cmp_fiber/pkg/constants"
"github.com/break/junhong_cmp_fiber/pkg/errors"
"github.com/break/junhong_cmp_fiber/pkg/queue"
"github.com/bytedance/sonic"
"github.com/hibiken/asynq"
"go.uber.org/zap"
)
// Service 同步服务
type Service struct {
queueClient *queue.Client
logger *zap.Logger
}
// NewService 创建同步服务实例
func NewService(queueClient *queue.Client, logger *zap.Logger) *Service {
return &Service{
queueClient: queueClient,
logger: logger,
}
}
// SyncSIMStatus 同步 SIM 卡状态(异步)
func (s *Service) SyncSIMStatus(ctx context.Context, iccids []string, forceSync bool) error {
// 构造任务载荷
payload := &task.SIMStatusSyncPayload{
RequestID: fmt.Sprintf("sim-sync-%d", getCurrentTimestamp()),
ICCIDs: iccids,
ForceSync: forceSync,
}
payloadBytes, err := sonic.Marshal(payload)
if err != nil {
s.logger.Error("序列化 SIM 状态同步任务载荷失败",
zap.Int("iccid_count", len(iccids)),
zap.Bool("force_sync", forceSync),
zap.Error(err))
return errors.Wrap(errors.CodeInternalError, err, "序列化 SIM 状态同步任务载荷失败")
}
// 提交任务到队列(高优先级)
err = s.queueClient.EnqueueTask(
ctx,
constants.TaskTypeSIMStatusSync,
payloadBytes,
asynq.Queue(constants.QueueCritical), // SIM 状态同步使用高优先级队列
asynq.MaxRetry(constants.DefaultRetryMax),
)
if err != nil {
s.logger.Error("提交 SIM 状态同步任务失败",
zap.Int("iccid_count", len(iccids)),
zap.Error(err))
return errors.Wrap(errors.CodeInternalError, err, "提交 SIM 状态同步任务失败")
}
s.logger.Info("SIM 状态同步任务已提交",
zap.Int("iccid_count", len(iccids)),
zap.Bool("force_sync", forceSync))
return nil
}
// SyncData 通用数据同步(异步)
func (s *Service) SyncData(ctx context.Context, syncType string, startDate string, endDate string, batchSize int) error {
// 设置默认批量大小
if batchSize <= 0 {
batchSize = 100 // 默认批量大小
}
// 构造任务载荷
payload := &task.DataSyncPayload{
RequestID: fmt.Sprintf("data-sync-%s-%d", syncType, getCurrentTimestamp()),
SyncType: syncType,
StartDate: startDate,
EndDate: endDate,
BatchSize: batchSize,
}
payloadBytes, err := sonic.Marshal(payload)
if err != nil {
s.logger.Error("序列化数据同步任务载荷失败",
zap.String("sync_type", syncType),
zap.String("start_date", startDate),
zap.String("end_date", endDate),
zap.Error(err))
return errors.Wrap(errors.CodeInternalError, err, "序列化数据同步任务载荷失败")
}
// 提交任务到队列(默认优先级)
err = s.queueClient.EnqueueTask(
ctx,
constants.TaskTypeDataSync,
payloadBytes,
asynq.Queue(constants.QueueDefault),
asynq.MaxRetry(constants.DefaultRetryMax),
)
if err != nil {
s.logger.Error("提交数据同步任务失败",
zap.String("sync_type", syncType),
zap.Error(err))
return errors.Wrap(errors.CodeInternalError, err, "提交数据同步任务失败")
}
s.logger.Info("数据同步任务已提交",
zap.String("sync_type", syncType),
zap.String("start_date", startDate),
zap.String("end_date", endDate),
zap.Int("batch_size", batchSize))
return nil
}
// SyncFlowUsage 同步流量使用数据(异步)
func (s *Service) SyncFlowUsage(ctx context.Context, startDate string, endDate string) error {
return s.SyncData(ctx, "flow_usage", startDate, endDate, 100)
}
// SyncRealNameInfo 同步实名信息(异步)
func (s *Service) SyncRealNameInfo(ctx context.Context, startDate string, endDate string) error {
return s.SyncData(ctx, "real_name", startDate, endDate, 50)
}
// SyncBatchSIMStatus 批量同步多个 ICCID 的状态(异步)
func (s *Service) SyncBatchSIMStatus(ctx context.Context, iccids []string) error {
// 如果 ICCID 列表为空,直接返回
if len(iccids) == 0 {
s.logger.Warn("批量同步 SIM 状态时 ICCID 列表为空")
return nil
}
// 分批处理(每批最多 100 个)
batchSize := 100
for i := 0; i < len(iccids); i += batchSize {
end := i + batchSize
if end > len(iccids) {
end = len(iccids)
}
batch := iccids[i:end]
if err := s.SyncSIMStatus(ctx, batch, false); err != nil {
s.logger.Error("批量同步 SIM 状态失败",
zap.Int("batch_start", i),
zap.Int("batch_end", end),
zap.Error(err))
return err
}
}
s.logger.Info("批量 SIM 状态同步任务已全部提交",
zap.Int("total_iccids", len(iccids)),
zap.Int("batch_size", batchSize))
return nil
}
// getCurrentTimestamp 获取当前时间戳(毫秒)
func getCurrentTimestamp() int64 {
return 0 // 实际实现应返回真实时间戳
}

View File

@@ -1,170 +0,0 @@
package task
import (
"context"
"fmt"
"time"
"github.com/bytedance/sonic"
"github.com/hibiken/asynq"
"github.com/redis/go-redis/v9"
"go.uber.org/zap"
"gorm.io/gorm"
"github.com/break/junhong_cmp_fiber/pkg/constants"
)
// SIMStatusSyncPayload SIM 卡状态同步任务载荷
type SIMStatusSyncPayload struct {
RequestID string `json:"request_id"`
ICCIDs []string `json:"iccids"` // ICCID 列表
ForceSync bool `json:"force_sync"` // 强制同步(忽略缓存)
}
// SIMHandler SIM 卡状态同步任务处理器
type SIMHandler struct {
db *gorm.DB
redis *redis.Client
logger *zap.Logger
}
// NewSIMHandler 创建 SIM 卡状态同步任务处理器
func NewSIMHandler(db *gorm.DB, redis *redis.Client, logger *zap.Logger) *SIMHandler {
return &SIMHandler{
db: db,
redis: redis,
logger: logger,
}
}
// HandleSIMStatusSync 处理 SIM 卡状态同步任务
func (h *SIMHandler) HandleSIMStatusSync(ctx context.Context, task *asynq.Task) error {
// 解析任务载荷
var payload SIMStatusSyncPayload
if err := sonic.Unmarshal(task.Payload(), &payload); err != nil {
h.logger.Error("解析 SIM 状态同步任务载荷失败",
zap.Error(err),
zap.String("task_id", task.ResultWriter().TaskID()),
)
return asynq.SkipRetry
}
// 验证载荷
if err := h.validatePayload(&payload); err != nil {
h.logger.Error("SIM 状态同步任务载荷验证失败",
zap.Error(err),
zap.String("request_id", payload.RequestID),
)
return asynq.SkipRetry
}
// 幂等性检查
lockKey := constants.RedisTaskLockKey(payload.RequestID)
locked, err := h.acquireLock(ctx, lockKey)
if err != nil {
h.logger.Error("获取任务锁失败",
zap.Error(err),
zap.String("request_id", payload.RequestID),
)
return err
}
if !locked {
h.logger.Info("任务已执行,跳过(幂等性)",
zap.String("request_id", payload.RequestID),
zap.Int("iccid_count", len(payload.ICCIDs)),
)
return nil
}
// 记录任务开始
h.logger.Info("开始处理 SIM 卡状态同步任务",
zap.String("request_id", payload.RequestID),
zap.Int("iccid_count", len(payload.ICCIDs)),
zap.Bool("force_sync", payload.ForceSync),
)
// 执行状态同步
if err := h.syncSIMStatus(ctx, &payload); err != nil {
h.logger.Error("SIM 卡状态同步失败",
zap.Error(err),
zap.String("request_id", payload.RequestID),
)
return err
}
// 记录任务完成
h.logger.Info("SIM 卡状态同步成功",
zap.String("request_id", payload.RequestID),
zap.Int("iccid_count", len(payload.ICCIDs)),
)
return nil
}
// validatePayload 验证 SIM 状态同步载荷
func (h *SIMHandler) validatePayload(payload *SIMStatusSyncPayload) error {
if payload.RequestID == "" {
return fmt.Errorf("request_id 不能为空")
}
if len(payload.ICCIDs) == 0 {
return fmt.Errorf("iccids 不能为空")
}
if len(payload.ICCIDs) > 1000 {
return fmt.Errorf("单次同步 ICCID 数量不能超过 1000")
}
return nil
}
// acquireLock 获取 Redis 锁
func (h *SIMHandler) acquireLock(ctx context.Context, key string) (bool, error) {
result, err := h.redis.SetNX(ctx, key, "1", 24*time.Hour).Result()
if err != nil {
return false, fmt.Errorf("设置 Redis 锁失败: %w", err)
}
return result, nil
}
// syncSIMStatus 执行 SIM 卡状态同步
func (h *SIMHandler) syncSIMStatus(ctx context.Context, payload *SIMStatusSyncPayload) error {
// TODO: 实际实现中需要调用运营商 API 获取 SIM 卡状态
// 批量处理 ICCID
batchSize := 100
for i := 0; i < len(payload.ICCIDs); i += batchSize {
// 检查上下文是否已取消
select {
case <-ctx.Done():
return ctx.Err()
default:
}
end := i + batchSize
if end > len(payload.ICCIDs) {
end = len(payload.ICCIDs)
}
batch := payload.ICCIDs[i:end]
h.logger.Debug("同步 SIM 卡状态批次",
zap.Int("batch_start", i),
zap.Int("batch_end", end),
zap.Int("batch_size", len(batch)),
)
// 模拟调用外部 API
time.Sleep(200 * time.Millisecond)
// TODO: 实际实现中需要:
// 1. 调用运营商 API 获取状态
// 2. 使用事务批量更新数据库
// 3. 更新 Redis 缓存
// 4. 记录同步日志
}
h.logger.Info("SIM 卡状态批量同步完成",
zap.Int("total_iccids", len(payload.ICCIDs)),
zap.Int("batch_size", batchSize),
)
return nil
}