From 65e461eff713f3ba8ec6dc4041658df6369e2323 Mon Sep 17 00:00:00 2001 From: huang Date: Sat, 28 Mar 2026 15:27:59 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BB=A3=E7=A0=81=E8=B4=A8=E9=87=8F?= =?UTF-8?q?=E6=B8=85=E7=90=86=20=E2=80=94=20=E7=BB=9F=E4=B8=80C=E7=AB=AF?= =?UTF-8?q?=E6=94=AF=E4=BB=98=E6=9E=9A=E4=B8=BE=E3=80=81=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E9=9D=99=E9=BB=98=E9=94=99=E8=AF=AF=E3=80=81=E6=B3=A8=E5=85=A5?= =?UTF-8?q?Worker=E5=9B=9E=E8=B0=83=E3=80=81=E7=A7=BB=E9=99=A4=E5=BA=9F?= =?UTF-8?q?=E5=BC=83=E5=90=8C=E6=AD=A5=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - C端订单支付状态直接使用管理端统一枚举(1/2/3/4)，移除冗余映射函数 - 资产查询绑定卡/套餐查询失败时记录Warn日志而非静默忽略 - Worker bootstrap注入停复机回调（流量耗尽停机、套餐激活复机） - 删除废弃的SIM状态同步服务和任务处理器 - 新增开发环境数据清理脚本（full/soft/table三种模式） --- internal/bootstrap/worker_services.go | 6 + internal/handler/app/client_order.go | 36 +- internal/model/dto/client_order_dto.go | 8 +- internal/service/asset/service.go | 21 +- internal/service/client_order/service.go | 15 +- internal/service/sync/service.go | 170 ------ internal/task/sim.go | 170 ------ pkg/constants/constants.go | 1 - pkg/queue/handler.go | 4 - scripts/cleanup/main.go | 636 +++++++++++++++++++++++ 10 files changed, 668 insertions(+), 399 deletions(-) delete mode 100644 internal/service/sync/service.go delete mode 100644 internal/task/sim.go create mode 100644 scripts/cleanup/main.go diff --git a/internal/bootstrap/worker_services.go b/internal/bootstrap/worker_services.go index e2feaa1..b4c1e60 100644 --- a/internal/bootstrap/worker_services.go +++ b/internal/bootstrap/worker_services.go @@ -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, diff --git a/internal/handler/app/client_order.go b/internal/handler/app/client_order.go index d742489..2fbc98a 100644 --- a/internal/handler/app/client_order.go +++ b/internal/handler/app/client_order.go @@ -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 "" diff --git a/internal/model/dto/client_order_dto.go b/internal/model/dto/client_order_dto.go index 6d2163d..4abf644 100644 --- a/internal/model/dto/client_order_dto.go +++ b/internal/model/dto/client_order_dto.go @@ -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:"支付时间"` diff --git a/internal/service/asset/service.go b/internal/service/asset/service.go index 9bdc977..1b715ec 100644 --- a/internal/service/asset/service.go +++ b/internal/service/asset/service.go @@ -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 diff --git a/internal/service/client_order/service.go b/internal/service/client_order/service.go index 8cd3944..d8030b9 100644 --- a/internal/service/client_order/service.go +++ b/internal/service/client_order/service.go @@ -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: diff --git a/internal/service/sync/service.go b/internal/service/sync/service.go deleted file mode 100644 index b9d8c60..0000000 --- a/internal/service/sync/service.go +++ /dev/null @@ -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 // 实际实现应返回真实时间戳 -} diff --git a/internal/task/sim.go b/internal/task/sim.go deleted file mode 100644 index c69e55e..0000000 --- a/internal/task/sim.go +++ /dev/null @@ -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 -} diff --git a/pkg/constants/constants.go b/pkg/constants/constants.go index 9bab15a..89ba877 100644 --- a/pkg/constants/constants.go +++ b/pkg/constants/constants.go @@ -44,7 +44,6 @@ const ( const ( TaskTypeEmailSend = "email:send" // 发送邮件 TaskTypeDataSync = "data:sync" // 数据同步 - TaskTypeSIMStatusSync = "sim:status:sync" // SIM 卡状态同步 TaskTypeCommission = "commission:calculate" // 分佣计算 TaskTypeIotCardImport = "iot_card:import" // IoT 卡批量导入 TaskTypeDeviceImport = "device:import" // 设备批量导入 diff --git a/pkg/queue/handler.go b/pkg/queue/handler.go index 115e64f..829202e 100644 --- a/pkg/queue/handler.go +++ b/pkg/queue/handler.go @@ -51,7 +51,6 @@ func NewHandler( func (h *Handler) RegisterHandlers() *asynq.ServeMux { emailHandler := task.NewEmailHandler(h.redis, h.logger) syncHandler := task.NewSyncHandler(h.db, h.logger) - simHandler := task.NewSIMHandler(h.db, h.redis, h.logger) h.mux.HandleFunc(constants.TaskTypeEmailSend, emailHandler.HandleEmailSend) h.logger.Info("注册邮件发送任务处理器", zap.String("task_type", constants.TaskTypeEmailSend)) @@ -59,9 +58,6 @@ func (h *Handler) RegisterHandlers() *asynq.ServeMux { h.mux.HandleFunc(constants.TaskTypeDataSync, syncHandler.HandleDataSync) h.logger.Info("注册数据同步任务处理器", zap.String("task_type", constants.TaskTypeDataSync)) - h.mux.HandleFunc(constants.TaskTypeSIMStatusSync, simHandler.HandleSIMStatusSync) - h.logger.Info("注册 SIM 状态同步任务处理器", zap.String("task_type", constants.TaskTypeSIMStatusSync)) - h.registerIotCardImportHandler() h.registerDeviceImportHandler() h.registerCommissionStatsHandlers() diff --git a/scripts/cleanup/main.go b/scripts/cleanup/main.go new file mode 100644 index 0000000..07d2e1d --- /dev/null +++ b/scripts/cleanup/main.go @@ -0,0 +1,636 @@ +//go:build ignore + +// 开发环境数据清理脚本 +// +// ⚠️ 警告：此脚本会删除数据，请确认连接的是开发/测试数据库！ +// +// 三种清理模式： +// full — 完全清理：只保留系统配置 + 超管账号 + 权限 + 运营商 +// soft — 轻量清理：只清流水/日志数据，保留组织架构、卡、设备、套餐 +// table — 指定清理：按表名清理指定的表 +// +// 用法： +// source .env.local +// go run ./scripts/cleanup/main.go -mode full # 完全清理 +// go run ./scripts/cleanup/main.go -mode soft # 轻量清理 +// go run ./scripts/cleanup/main.go -mode table -tables "tb_order,tb_order_item" +// go run ./scripts/cleanup/main.go -mode full -yes # 跳过确认 +// go run ./scripts/cleanup/main.go -mode full -dry-run # 只预览不执行 + +package main + +import ( + "bufio" + "flag" + "fmt" + "log" + "os" + "strings" + "time" + + "gorm.io/driver/postgres" + "gorm.io/gorm" + "gorm.io/gorm/logger" +) + +// ==================== 表分类定义 ==================== + +// 系统配置表 — 任何模式都不清理 +var systemTables = []string{ + "schema_migrations", + "tb_permission", + "tb_carrier", + "tb_polling_config", + "tb_polling_concurrency_config", + "tb_polling_alert_rule", + "tb_data_cleanup_config", + "tb_wechat_config", + "tb_payment_merchant_setting", + "tb_commission_withdrawal_setting", + "tb_dev_capability_config", +} + +// 事务/流水/日志表 — soft 和 full 模式都清理（TRUNCATE） +var transactionalTables = []string{ + // 流量记录（最大的表，优先清理释放空间） + "tb_data_usage_record", + "tb_card_daily_usage", + // 订单 + "tb_order_item", + "tb_order", + // 套餐使用 + "tb_package_usage_daily_record", + "tb_package_usage", + // 资产钱包 + "tb_asset_wallet_transaction", + "tb_asset_recharge_record", + "tb_asset_allocation_record", + "tb_asset_wallet", + // 代理钱包 + "tb_agent_wallet_transaction", + "tb_agent_recharge_record", + "tb_agent_wallet", + // 佣金 + "tb_commission_record", + "tb_commission_withdrawal_request", + "tb_shop_series_commission_stats", + // 换卡/换货 + "tb_card_replacement_record_legacy", + "tb_card_replacement_request", + "tb_exchange_order", + // 个人客户（先删关联表再删主表） + "tb_personal_customer_device", + "tb_personal_customer_iccid", + "tb_personal_customer_phone", + "tb_personal_customer_openid", + "tb_personal_customer", + // 遗留个人客户表（无 tb_ 前缀） + "personal_customer_device", + "personal_customer_iccid", + "personal_customer_phone", + // 操作日志 + "tb_account_operation_log", + "tb_data_cleanup_log", + "tb_polling_alert_history", + "tb_polling_manual_trigger_log", +} + +// 业务数据表 — 仅 full 模式清理（TRUNCATE） +var businessTables = []string{ + // 授权 + "tb_enterprise_card_authorization", + "tb_enterprise_device_authorization", + // 设备 + "tb_device_sim_binding", + "tb_device", + "tb_device_import_task", + // IoT 卡 + "tb_iot_card", + "tb_iot_card_import_task", + "tb_number_card", + // 套餐分配 + "tb_shop_package_allocation_price_history", + "tb_shop_package_allocation", + "tb_shop_series_allocation", + // 套餐定义 + "tb_package", + "tb_package_series", + // 标签 + "tb_resource_tag", + "tb_tag", + // 组织架构 + "tb_shop_role", + "tb_enterprise", + "tb_shop", +} + +// 部分清理规则 — full 模式下保留关键行（DELETE WHERE） +// 按执行顺序排列：先删关联数据，再删主数据 +type partialCleanup struct { + table string // 表名 + keepWhere string // 保留满足此条件的行 + desc string // 保留说明 +} + +var partialCleanupRules = []partialCleanup{ + { + table: "tb_account_role", + keepWhere: "account_id IN (SELECT id FROM tb_account WHERE user_type = 1)", + desc: "超管角色绑定", + }, + { + table: "tb_role_permission", + keepWhere: "role_id IN (SELECT id FROM tb_role WHERE role_name = '超级管理员')", + desc: "超管权限映射", + }, + { + table: "tb_role", + keepWhere: "role_name = '超级管理员'", + desc: "超级管理员角色", + }, + { + table: "tb_account", + keepWhere: "user_type = 1", + desc: "超级管理员账号", + }, +} + +// ==================== 主入口 ==================== + +var ( + flagMode = flag.String("mode", "", "清理模式: full / soft / table") + flagTables = flag.String("tables", "", "table 模式: 指定表名（逗号分隔）") + flagYes = flag.Bool("yes", false, "跳过确认直接执行") + flagDryRun = flag.Bool("dry-run", false, "预览模式，只显示将要清理的内容") +) + +func main() { + flag.Parse() + + if *flagMode == "" { + showUsage() + os.Exit(0) + } + + db := connectDB() + + switch *flagMode { + case "full": + doFull(db) + case "soft": + doSoft(db) + case "table": + if *flagTables == "" { + log.Fatal("❌ table 模式必须指定 -tables 参数，例如: -tables \"tb_order,tb_order_item\"") + } + doTable(db) + default: + log.Fatalf("❌ 未知模式: %s（支持: full / soft / table）", *flagMode) + } +} + +// ==================== 清理模式实现 ==================== + +// doFull 完全清理：只保留系统配置 + 超管 +func doFull(db *gorm.DB) { + fmt.Println("\n🔴 模式: full — 完全清理") + fmt.Println(" 保留: 系统配置表 + 超管账号 + 超管角色/权限 + 运营商") + fmt.Println(" 清理: 所有业务数据 + 组织架构 + 卡/设备 + 套餐 + 流水/日志") + + // 收集所有要清理的表 + truncateList := append([]string{}, transactionalTables...) + truncateList = append(truncateList, businessTables...) + + // 显示影响概览 + fmt.Println("\n📊 数据影响预览:") + showTableStats(db, "🗑️ TRUNCATE（完全清空）", truncateList) + showPartialStats(db, "✂️ DELETE（部分保留）", partialCleanupRules) + showTableStats(db, "🔒 保留（不动）", systemTables) + + if *flagDryRun { + fmt.Println("\n⏸️ dry-run 模式，未执行任何操作") + return + } + + if !confirmAction("full 清理") { + return + } + + start := time.Now() + + // 第一步：TRUNCATE 事务表 + 业务表 + fmt.Println("\n⏳ 步骤 1/3: TRUNCATE 事务和业务数据表...") + existingTables := filterExistingTables(db, truncateList) + if len(existingTables) > 0 { + truncated := truncateTables(db, existingTables) + fmt.Printf(" ✅ 清空了 %d 张表\n", truncated) + } + + // 第二步：DELETE 部分保留表 + fmt.Println("\n⏳ 步骤 2/3: 清理部分保留表...") + for _, rule := range partialCleanupRules { + deleted := deletePartial(db, rule) + fmt.Printf(" ✅ %s: 删除 %d 行（保留: %s）\n", rule.table, deleted, rule.desc) + } + + // 第三步：重置序列 + fmt.Println("\n⏳ 步骤 3/3: 重置部分保留表的序列...") + resetPartialSequences(db) + + elapsed := time.Since(start) + fmt.Printf("\n🎉 full 清理完成！耗时: %v\n", elapsed.Round(time.Millisecond)) + showPostCleanupTips() +} + +// doSoft 轻量清理：只清流水数据 +func doSoft(db *gorm.DB) { + fmt.Println("\n🟡 模式: soft — 轻量清理") + fmt.Println(" 保留: 系统配置 + 所有账号 + 组织架构 + 卡/设备 + 套餐") + fmt.Println(" 清理: 订单 + 钱包流水 + 佣金记录 + 操作日志 + 流量记录") + + fmt.Println("\n📊 数据影响预览:") + showTableStats(db, "🗑️ TRUNCATE（完全清空）", transactionalTables) + kept := append([]string{}, businessTables...) + kept = append(kept, systemTables...) + for _, rule := range partialCleanupRules { + kept = append(kept, rule.table) + } + showTableStats(db, "🔒 保留（不动）", kept) + + if *flagDryRun { + fmt.Println("\n⏸️ dry-run 模式，未执行任何操作") + return + } + + if !confirmAction("soft 清理") { + return + } + + start := time.Now() + + fmt.Println("\n⏳ TRUNCATE 事务/流水表...") + existingTables := filterExistingTables(db, transactionalTables) + if len(existingTables) > 0 { + truncated := truncateTables(db, existingTables) + fmt.Printf(" ✅ 清空了 %d 张表\n", truncated) + } + + elapsed := time.Since(start) + fmt.Printf("\n🎉 soft 清理完成！耗时: %v\n", elapsed.Round(time.Millisecond)) + showPostCleanupTips() +} + +// doTable 按表名清理 +func doTable(db *gorm.DB) { + tableNames := parseTableNames(*flagTables) + if len(tableNames) == 0 { + log.Fatal("❌ 未指定有效的表名") + } + + fmt.Println("\n🔵 模式: table — 指定表清理") + fmt.Printf(" 目标表: %s\n", strings.Join(tableNames, ", ")) + + // 检查是否包含系统表 + systemSet := toSet(systemTables) + var warnings []string + var safeTables []string + for _, t := range tableNames { + if _, ok := systemSet[t]; ok { + warnings = append(warnings, t) + } else { + safeTables = append(safeTables, t) + } + } + if len(warnings) > 0 { + fmt.Printf("\n ⚠️ 以下系统表已跳过（不允许清理）: %s\n", strings.Join(warnings, ", ")) + } + if len(safeTables) == 0 { + fmt.Println("\n 没有可清理的表") + return + } + + // 检查表是否存在 + existingTables := filterExistingTables(db, safeTables) + if len(existingTables) == 0 { + fmt.Println("\n 指定的表在数据库中均不存在") + return + } + if len(existingTables) < len(safeTables) { + missing := diff(safeTables, existingTables) + fmt.Printf(" ⚠️ 以下表不存在，已跳过: %s\n", strings.Join(missing, ", ")) + } + + fmt.Println("\n📊 数据影响预览:") + showTableStats(db, "🗑️ TRUNCATE（完全清空）", existingTables) + + if *flagDryRun { + fmt.Println("\n⏸️ dry-run 模式，未执行任何操作") + return + } + + if !confirmAction("table 清理") { + return + } + + start := time.Now() + + fmt.Println("\n⏳ TRUNCATE 指定表...") + truncated := truncateTables(db, existingTables) + fmt.Printf(" ✅ 清空了 %d 张表\n", truncated) + + elapsed := time.Since(start) + fmt.Printf("\n🎉 table 清理完成！耗时: %v\n", elapsed.Round(time.Millisecond)) + showPostCleanupTips() +} + +// ==================== 执行辅助函数 ==================== + +// truncateTables 批量 TRUNCATE 多张表（重置序列，级联） +func truncateTables(db *gorm.DB, tables []string) int { + if len(tables) == 0 { + return 0 + } + + // PostgreSQL 支持单条 TRUNCATE 多张表，原子操作且更快 + sql := fmt.Sprintf("TRUNCATE TABLE %s RESTART IDENTITY CASCADE", strings.Join(tables, ", ")) + result := db.Exec(sql) + if result.Error != nil { + // 如果批量失败，尝试逐表清理 + fmt.Printf(" ⚠️ 批量 TRUNCATE 失败: %v，改为逐表清理...\n", result.Error) + count := 0 + for _, t := range tables { + err := db.Exec(fmt.Sprintf("TRUNCATE TABLE %s RESTART IDENTITY CASCADE", t)).Error + if err != nil { + fmt.Printf(" ⚠️ 跳过 %s: %v\n", t, err) + } else { + count++ + } + } + return count + } + return len(tables) +} + +// deletePartial 删除表中不满足保留条件的行 +func deletePartial(db *gorm.DB, rule partialCleanup) int64 { + sql := fmt.Sprintf("DELETE FROM %s WHERE NOT (%s)", rule.table, rule.keepWhere) + result := db.Exec(sql) + if result.Error != nil { + fmt.Printf(" ⚠️ %s 清理失败: %v\n", rule.table, result.Error) + return 0 + } + return result.RowsAffected +} + +// resetPartialSequences 重置部分保留表的序列到当前最大 ID + 1 +func resetPartialSequences(db *gorm.DB) { + for _, rule := range partialCleanupRules { + // 查找表对应的序列名 + var seqName string + db.Raw(` + SELECT pg_get_serial_sequence(?, 'id') + `, rule.table).Scan(&seqName) + + if seqName == "" { + continue + } + + // 获取当前最大 ID + var maxID *int64 + db.Raw(fmt.Sprintf("SELECT MAX(id) FROM %s", rule.table)).Scan(&maxID) + + if maxID == nil || *maxID == 0 { + // 表为空，重置到 1 + db.Exec(fmt.Sprintf("ALTER SEQUENCE %s RESTART WITH 1", seqName)) + } else { + // 重置到 max + 1 + db.Exec(fmt.Sprintf("SELECT setval('%s', %d)", seqName, *maxID)) + } + fmt.Printf(" ✅ %s 序列已重置\n", rule.table) + } +} + +// filterExistingTables 过滤出数据库中实际存在的表 +func filterExistingTables(db *gorm.DB, tables []string) []string { + var existing []string + for _, t := range tables { + var count int64 + db.Raw("SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = 'public' AND table_name = ?", t).Scan(&count) + if count > 0 { + existing = append(existing, t) + } + } + return existing +} + +// ==================== 显示辅助函数 ==================== + +// showTableStats 显示一组表的行数统计 +func showTableStats(db *gorm.DB, title string, tables []string) { + fmt.Printf("\n %s:\n", title) + totalRows := int64(0) + for _, t := range tables { + rows := getEstimatedRows(db, t) + totalRows += rows + if rows > 0 { + fmt.Printf(" %-50s %s 行\n", t, formatNumber(rows)) + } else { + fmt.Printf(" %-50s (空)\n", t) + } + } + if totalRows > 0 { + fmt.Printf(" %s\n", strings.Repeat("─", 65)) + fmt.Printf(" %-50s %s 行\n", "合计", formatNumber(totalRows)) + } +} + +// showPartialStats 显示部分清理表的统计 +func showPartialStats(db *gorm.DB, title string, rules []partialCleanup) { + fmt.Printf("\n %s:\n", title) + for _, rule := range rules { + total := getEstimatedRows(db, rule.table) + // 查询将被保留的行数 + var keepCount int64 + db.Raw(fmt.Sprintf("SELECT COUNT(*) FROM %s WHERE %s", rule.table, rule.keepWhere)).Scan(&keepCount) + deleteCount := total - keepCount + if deleteCount < 0 { + deleteCount = 0 + } + fmt.Printf(" %-50s 删除 %s / 保留 %s（%s）\n", + rule.table, formatNumber(deleteCount), formatNumber(keepCount), rule.desc) + } +} + +// showUsage 显示使用说明 +func showUsage() { + fmt.Println(` +📋 开发环境数据清理脚本 + +用法: + source .env.local + go run ./scripts/cleanup/main.go [选项] + +模式: + -mode full 完全清理 — 只保留系统配置 + 超管账号 + 适用场景：重新开始，从零开始测试全流程 + + -mode soft 轻量清理 — 只清流水数据 + 适用场景：保留卡/设备/套餐，重新测试订单/支付流程 + + -mode table 指定清理 — 按表名清理 + 配合 -tables "tb_order,tb_order_item" 使用 + +选项: + -yes 跳过确认，直接执行 + -dry-run 预览模式，只显示将要清理的内容，不实际执行 + -tables table 模式下指定表名（逗号分隔） + +示例: + go run ./scripts/cleanup/main.go -mode full # 完全清理 + go run ./scripts/cleanup/main.go -mode soft -yes # 轻量清理（跳过确认） + go run ./scripts/cleanup/main.go -mode full -dry-run # 预览 full 清理 + go run ./scripts/cleanup/main.go -mode table -tables "tb_order,tb_order_item,tb_data_usage_record"`) +} + +// showPostCleanupTips 清理后提示 +func showPostCleanupTips() { + fmt.Println("\n💡 后续建议:") + fmt.Println(" 1. 执行 VACUUM ANALYZE 回收磁盘空间（大表清理后建议执行）:") + fmt.Println(" psql -c 'VACUUM ANALYZE;'") + fmt.Println(" 2. 清理 Redis 缓存（店铺层级、权限缓存等）:") + fmt.Println(" redis-cli FLUSHDB") + fmt.Println(" 3. 重启 API/Worker 服务以刷新内存缓存") +} + +// ==================== 数据库连接 ==================== + +func connectDB() *gorm.DB { + dsn := buildDSN() + + db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{ + Logger: logger.Default.LogMode(logger.Warn), + }) + if err != nil { + log.Fatalf("❌ 连接数据库失败: %v", err) + } + + sqlDB, _ := db.DB() + sqlDB.SetMaxOpenConns(5) + sqlDB.SetMaxIdleConns(3) + + // 获取数据库名用于安全确认 + var dbName string + db.Raw("SELECT current_database()").Scan(&dbName) + fmt.Printf("✅ 已连接数据库: %s\n", dbName) + + return db +} + +func buildDSN() string { + host := getEnv("JUNHONG_DATABASE_HOST", "") + port := getEnv("JUNHONG_DATABASE_PORT", "5432") + user := getEnv("JUNHONG_DATABASE_USER", "") + password := getEnv("JUNHONG_DATABASE_PASSWORD", "") + dbname := getEnv("JUNHONG_DATABASE_DBNAME", "") + + if host == "" || user == "" || dbname == "" { + log.Fatal("❌ 缺少数据库配置，请设置环境变量:\n" + + " export JUNHONG_DATABASE_HOST=...\n" + + " export JUNHONG_DATABASE_USER=...\n" + + " export JUNHONG_DATABASE_PASSWORD=...\n" + + " export JUNHONG_DATABASE_DBNAME=...\n" + + " 或执行: source .env.local") + } + + return fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable TimeZone=Asia/Shanghai", + host, port, user, password, dbname) +} + +// ==================== 工具函数 ==================== + +// confirmAction 确认操作（-yes 跳过） +func confirmAction(action string) bool { + if *flagYes { + return true + } + + var dbName string + fmt.Printf("\n⚠️ 即将对数据库执行 [%s]，此操作不可撤销！\n", action) + fmt.Print(" 输入 yes 确认执行: ") + + scanner := bufio.NewScanner(os.Stdin) + if scanner.Scan() { + dbName = strings.TrimSpace(scanner.Text()) + } + + if dbName != "yes" { + fmt.Println(" ❌ 已取消") + return false + } + return true +} + +// getEstimatedRows 获取表的估算行数（从 pg_stat 读取，非常快） +func getEstimatedRows(db *gorm.DB, tableName string) int64 { + var rows int64 + db.Raw("SELECT COALESCE(n_live_tup, 0) FROM pg_stat_user_tables WHERE relname = ?", tableName).Scan(&rows) + return rows +} + +// formatNumber 千分位格式化数字 +func formatNumber(n int64) string { + if n < 1000 { + return fmt.Sprintf("%d", n) + } + s := fmt.Sprintf("%d", n) + result := make([]byte, 0, len(s)+(len(s)-1)/3) + for i, c := range s { + if i > 0 && (len(s)-i)%3 == 0 { + result = append(result, ',') + } + result = append(result, byte(c)) + } + return string(result) +} + +// parseTableNames 解析逗号分隔的表名 +func parseTableNames(input string) []string { + parts := strings.Split(input, ",") + var result []string + for _, p := range parts { + t := strings.TrimSpace(p) + if t != "" { + result = append(result, t) + } + } + return result +} + +// toSet 将字符串切片转为 map +func toSet(items []string) map[string]struct{} { + m := make(map[string]struct{}, len(items)) + for _, item := range items { + m[item] = struct{}{} + } + return m +} + +// diff 返回在 a 中但不在 b 中的元素 +func diff(a, b []string) []string { + bSet := toSet(b) + var result []string + for _, item := range a { + if _, ok := bSet[item]; !ok { + result = append(result, item) + } + } + return result +} + +// getEnv 获取环境变量，支持默认值 +func getEnv(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + return fallback +}