diff --git a/cmd/api/docs.go b/cmd/api/docs.go index 5bcfff1c..41c19b3a 100644 --- a/cmd/api/docs.go +++ b/cmd/api/docs.go @@ -39,6 +39,7 @@ func generateOpenAPIDocs(outputPath string, logger *zap.Logger) { handlers.AssetAutoRenewal = admin.NewAssetAutoRenewalConfigHandler(nil, nil) // 运营报表 Handler 必须同时进入文档工厂，避免新增管理接口遗漏文档注册。 handlers.OperationsReport = admin.NewOperationsReportHandler(nil, nil, nil) + handlers.StopResumeRule = admin.NewStopResumeRuleHandler(nil, nil, nil) handlers.ClientPopup = apphandler.NewClientPopupHandler(nil, nil, nil) handlers.H5PopupConfiguration = admin.NewH5PopupConfigurationHandler(nil, nil, nil) // 企业微信 Handler 在此显式装配，避免新增管理接口遗漏文档注册。 diff --git a/cmd/gendocs/main.go b/cmd/gendocs/main.go index aab1ae7f..4578428a 100644 --- a/cmd/gendocs/main.go +++ b/cmd/gendocs/main.go @@ -48,6 +48,7 @@ func generateAdminDocs(outputPath string) error { handlers.AssetAutoRenewal = admin.NewAssetAutoRenewalConfigHandler(nil, nil) // 运营报表 Handler 必须同时进入文档工厂，避免新增管理接口遗漏文档注册。 handlers.OperationsReport = admin.NewOperationsReportHandler(nil, nil, nil) + handlers.StopResumeRule = admin.NewStopResumeRuleHandler(nil, nil, nil) handlers.ClientPopup = apphandler.NewClientPopupHandler(nil, nil, nil) handlers.H5PopupConfiguration = admin.NewH5PopupConfigurationHandler(nil, nil, nil) // 企业微信 Handler 在此显式装配，避免新增管理接口遗漏文档注册。 diff --git a/internal/application/cardobservation/network.go b/internal/application/cardobservation/network.go index 6ad64934..c3e132c5 100644 --- a/internal/application/cardobservation/network.go +++ b/internal/application/cardobservation/network.go @@ -46,7 +46,7 @@ func (s *Service) ApplyNetworkObservation(ctx context.Context, observation domai auditedCard = &card nextDecision, decisionErr := domain.ApplyNetwork(domain.CardNetworkSnapshot{ CardID: card.ID, NetworkStatus: card.NetworkStatus, StopReason: card.StopReason, - IsStandalone: card.IsStandalone, EnablePolling: card.EnablePolling, + EnablePolling: card.EnablePolling, }, observation) if decisionErr != nil { return decisionErr diff --git a/internal/bootstrap/handlers.go b/internal/bootstrap/handlers.go index 82aef063..876caf30 100644 --- a/internal/bootstrap/handlers.go +++ b/internal/bootstrap/handlers.go @@ -44,6 +44,7 @@ import ( packagetrafficalertquery "github.com/break/junhong_cmp_fiber/internal/query/packagetrafficalert" priorityPollingQuery "github.com/break/junhong_cmp_fiber/internal/query/prioritypolling" shopQuery "github.com/break/junhong_cmp_fiber/internal/query/shop" + "github.com/break/junhong_cmp_fiber/internal/query/stopresumerule" systemConfigQuery "github.com/break/junhong_cmp_fiber/internal/query/systemconfig" clientOrderSvc "github.com/break/junhong_cmp_fiber/internal/service/client_order" "github.com/break/junhong_cmp_fiber/internal/service/paymentmethod" @@ -303,6 +304,7 @@ func initHandlers(svc *services, deps *Dependencies) *Handlers { PackageTrafficAlert: admin.NewPackageTrafficAlertHandler(svc.PackageTrafficAlertRule, packageTrafficAlertQuery, svc.ExportTask, validate), AssetAutoRenewal: admin.NewAssetAutoRenewalConfigHandler(svc.AssetAutoRenewal, validate), OperationsReport: admin.NewOperationsReportHandler(operationsReportQuery, svc.ExportTask, validate), + StopResumeRule: admin.NewStopResumeRuleHandler(stopresumerule.NewQuery(deps.DB), svc.ExportTask, validate), ShopPackageBatchAllocation: admin.NewShopPackageBatchAllocationHandler(svc.ShopPackageBatchAllocation), ShopPackageBatchPricing: admin.NewShopPackageBatchPricingHandler(svc.ShopPackageBatchPricing), ShopSeriesGrant: admin.NewShopSeriesGrantHandler(svc.ShopSeriesGrant), diff --git a/internal/bootstrap/services.go b/internal/bootstrap/services.go index f8b5b267..4db09d8c 100644 --- a/internal/bootstrap/services.go +++ b/internal/bootstrap/services.go @@ -270,6 +270,8 @@ func initServices(s *stores, deps *Dependencies) *services { stopResumeService.SetUnifiedAudit(auditWriter, integrationlog.NewRepository(deps.DB)) if deps.GatewayLimit != nil { stopResumeService.SetStopResumeRecorder(deps.GatewayLimit.Recorder) + deps.GatewayLimit.TerminalPolling.SetAuditWriter(auditWriter) + stopResumeService.SetTerminalPollingDisabler(deps.GatewayLimit.TerminalPolling) } // 持通道阈值锁的卡在周期内拒绝一切复机（自动、手动、机卡分离）。 stopResumeService.SetChannelThresholdLockGuard(carrierThresholdService) @@ -303,6 +305,7 @@ func initServices(s *stores, deps *Dependencies) *services { device.SetObservationSeriesDispatcher(observationSeries) if deps.GatewayLimit != nil { device.SetStopResumeRecorder(deps.GatewayLimit.Recorder) + device.SetTerminalPollingDisabler(deps.GatewayLimit.TerminalPolling) } operationPassword := operationPasswordSvc.New(deps.Redis) shopCommission := shopCommissionSvc.New(s.Shop, s.Account, s.AgentWallet, s.CommissionWithdrawalRequest, s.CommissionWithdrawalSetting, s.CommissionRecord, s.AgentWalletTransaction, deps.DB, deps.Logger) diff --git a/internal/bootstrap/types.go b/internal/bootstrap/types.go index 0ddd4aaa..dc495112 100644 --- a/internal/bootstrap/types.go +++ b/internal/bootstrap/types.go @@ -87,6 +87,7 @@ type Handlers struct { PackageTrafficAlert *admin.PackageTrafficAlertHandler AssetAutoRenewal *admin.AssetAutoRenewalConfigHandler OperationsReport *admin.OperationsReportHandler + StopResumeRule *admin.StopResumeRuleHandler ClientWechat *app.ClientWechatHandler SuperAdmin *admin.SuperAdminHandler SystemConfig *admin.SystemConfigHandler diff --git a/internal/bootstrap/worker_services.go b/internal/bootstrap/worker_services.go index edf717f5..e9cf6ccf 100644 --- a/internal/bootstrap/worker_services.go +++ b/internal/bootstrap/worker_services.go @@ -194,6 +194,8 @@ func initWorkerServices(stores *queue.WorkerStores, deps *WorkerDependencies) *q stopResumeService.SetUnifiedAudit(auditWriter, integrationlog.NewRepository(deps.DB)) if deps.GatewayLimit != nil { stopResumeService.SetStopResumeRecorder(deps.GatewayLimit.Recorder) + deps.GatewayLimit.TerminalPolling.SetAuditWriter(auditWriter) + stopResumeService.SetTerminalPollingDisabler(deps.GatewayLimit.TerminalPolling) } // 持通道阈值锁的卡在周期内拒绝一切复机（自动、手动、机卡分离）。 stopResumeService.SetChannelThresholdLockGuard(carrierThresholdService) diff --git a/internal/domain/cardobservation/network.go b/internal/domain/cardobservation/network.go index e6330b5d..2fd8a00a 100644 --- a/internal/domain/cardobservation/network.go +++ b/internal/domain/cardobservation/network.go @@ -21,7 +21,6 @@ type CardNetworkSnapshot struct { CardID uint NetworkStatus int StopReason string - IsStandalone bool EnablePolling bool } @@ -39,7 +38,7 @@ type NetworkDecision struct { StopPolling bool } -// ApplyNetwork 应用 Gateway 状态映射、运营商停机原因和独立风险卡规则。 +// ApplyNetwork 应用 Gateway 状态映射、运营商停机原因和风险终止卡关闭轮询规则。 func ApplyNetwork(snapshot CardNetworkSnapshot, observation NetworkObservation) (NetworkDecision, error) { if err := validateObservationMetadata(observation.CardID, observation.Metadata); err != nil { return NetworkDecision{}, err @@ -53,7 +52,7 @@ func ApplyNetwork(snapshot CardNetworkSnapshot, observation NetworkObservation) decision := NetworkDecision{ StatusKnown: known, BeforeStatus: snapshot.NetworkStatus, AfterStatus: snapshot.NetworkStatus, GatewayExtend: extend, GatewayIMEI: imei, UpdateIMEI: imei != "", StopReason: snapshot.StopReason, - StopPolling: snapshot.EnablePolling && ShouldStopPollingForRisk(snapshot.IsStandalone, extend), + StopPolling: snapshot.EnablePolling && ShouldStopPollingForRisk(extend), } if !known { return decision, nil @@ -84,7 +83,8 @@ func MapGatewayNetworkStatus(cardStatus, extend string) (int, bool) { return constants.NetworkStatusOffline, false } -// ShouldStopPollingForRisk 判断独立卡是否命中运营商风险终止状态。 -func ShouldStopPollingForRisk(isStandalone bool, extend string) bool { - return isStandalone && (extend == constants.GatewayCardExtendRiskStop || extend == constants.GatewayCardExtendCancelled) +// ShouldStopPollingForRisk 判断卡是否命中运营商风险终止状态（风险停机、已销户）。 +// 不论是否绑定设备都关闭该卡自身轮询，不级联设备与设备下其他卡。 +func ShouldStopPollingForRisk(extend string) bool { + return extend == constants.GatewayCardExtendRiskStop || extend == constants.GatewayCardExtendCancelled } diff --git a/internal/domain/stopresume/rule.go b/internal/domain/stopresume/rule.go new file mode 100644 index 00000000..3cc6ef0a --- /dev/null +++ b/internal/domain/stopresume/rule.go @@ -0,0 +1,137 @@ +// Package stopresume 定义停复机新规则的纯判断：输入卡与资产的当前事实，输出目标动作，不读写任何外部状态。 +// +// 新规则（finalize-unified-stop-resume-rule）： +// - 卡或其有效绑定设备暂停轮询时不自动停复机； +// - 开机卡命中任一停机条件即停机：无套餐、流量用完、通道开启实名管控且卡未实名； +// - 停机卡同时满足有套餐、流量未用完、实名条件时复机，不看停机原因；持本周期通道阈值锁的卡拒绝复机； +// - 卡所属通道不存在（无法判断实名管控）时本次不动作。 +package stopresume + +import "github.com/break/junhong_cmp_fiber/pkg/constants" + +// 判断结果。 +const ( + ResultWillResume = "will_resume" // 将复机 + ResultWillStop = "will_stop" // 将停机 + ResultNoAction = "no_action" // 不动作 +) + +// 不动作原因。 +const ( + NoActionPollingSuspended = "polling_suspended" // 卡或所在设备暂停轮询 + NoActionConditionsMet = "conditions_met" // 开机卡满足开机条件 + NoActionNoPackage = "no_package" // 停机卡无套餐（与停机原因同值，共用名称） + NoActionTrafficExhausted = "traffic_exhausted" // 停机卡流量已用完 + NoActionNotRealname = "not_realname" // 停机卡所属通道开启实名管控且卡未实名 + NoActionThresholdLocked = "threshold_locked" // 停机卡持本周期通道阈值锁 + NoActionCarrierMissing = "carrier_missing" // 卡所属通道不存在，无法判断实名管控 + NoActionUnknownNetwork = "unknown_network" // 卡网络状态既非开机也非停机 +) + +// Facts 是判断一张卡所需的全部事实；套餐与流量按绑定设备（有有效绑定时）或卡自身计算。 +type Facts struct { + NetworkStatus int + PollingSuspended bool // 卡自身或其有效绑定设备暂停轮询 + HasValidPackage bool // 存在生效中的套餐 + TrafficExhausted bool // 生效或已用完的套餐全部达到流量上限 + RealnameVerified bool + CarrierKnown bool // 卡所属通道存在 + RealnameControl bool // 卡所属通道开启实名管控（可为假设值） + ThresholdLocked bool // 持本周期通道阈值锁 + CurrentStopReason string +} + +// Decision 是一张卡的判断结果。 +type Decision struct { + Result string // ResultWillResume / ResultWillStop / ResultNoAction + Reason string // 将停机时为目标停机原因；不动作时为不动作原因；将复机时为空 + StopReason string // 将停机时的目标停机原因 +} + +// Decide 按新规则判断卡的目标动作。 +func Decide(f Facts) Decision { + if f.PollingSuspended { + return noAction(NoActionPollingSuspended) + } + if !f.CarrierKnown { + return noAction(NoActionCarrierMissing) + } + realnameOK := !f.RealnameControl || f.RealnameVerified + switch f.NetworkStatus { + case constants.NetworkStatusOnline: + // 停机原因优先级：流量用完 > 无套餐（仅在未用完时判断）> 未实名，与评估主流程一致。 + switch { + case f.TrafficExhausted: + return stop(constants.StopReasonTrafficExhausted) + case !f.HasValidPackage: + return stop(constants.StopReasonNoPackage) + case !realnameOK: + return stop(constants.StopReasonNotRealname) + default: + return noAction(NoActionConditionsMet) + } + case constants.NetworkStatusOffline: + switch { + case !f.HasValidPackage: + return noAction(NoActionNoPackage) + case f.TrafficExhausted: + return noAction(NoActionTrafficExhausted) + case !realnameOK: + return noAction(NoActionNotRealname) + case f.ThresholdLocked: + return noAction(NoActionThresholdLocked) + default: + return Decision{Result: ResultWillResume} + } + default: + return noAction(NoActionUnknownNetwork) + } +} + +func stop(reason string) Decision { + return Decision{Result: ResultWillStop, Reason: reason, StopReason: reason} +} + +func noAction(reason string) Decision { + return Decision{Result: ResultNoAction, Reason: reason} +} + +// ResultName 返回判断结果中文名称。 +func ResultName(result string) string { + switch result { + case ResultWillResume: + return "将复机" + case ResultWillStop: + return "将停机" + case ResultNoAction: + return "不动作" + default: + return result + } +} + +// ReasonName 返回停机原因或不动作原因的中文名称。 +func ReasonName(reason string) string { + switch reason { + case "": + return "" + case NoActionPollingSuspended: + return "暂停轮询" + case NoActionConditionsMet: + return "满足开机条件" + case NoActionNoPackage: + return "无套餐" + case NoActionTrafficExhausted: + return "流量已用完" + case NoActionNotRealname: + return "未实名（通道开启实名管控）" + case NoActionThresholdLocked: + return "持通道阈值锁" + case NoActionCarrierMissing: + return "所属通道不存在" + case NoActionUnknownNetwork: + return "网络状态未知" + default: + return reason + } +} diff --git a/internal/exporter/registry.go b/internal/exporter/registry.go index 5779df5e..965d16a9 100644 --- a/internal/exporter/registry.go +++ b/internal/exporter/registry.go @@ -41,6 +41,7 @@ func NewDefaultRegistry(db *gorm.DB) *Registry { NewExpiringAssetDataSource(db), NewOperationsActivationDataSource(db), NewOperationsRenewalDataSource(db), + NewStopResumeRuleDataSource(db), ) } @@ -81,7 +82,8 @@ func IsSupportedScene(scene string) bool { constants.ExportTaskScenePackageTrafficAlert, constants.ExportTaskSceneExpiringAsset, constants.ExportTaskSceneOperationsActivation, - constants.ExportTaskSceneOperationsRenewal: + constants.ExportTaskSceneOperationsRenewal, + constants.ExportTaskSceneStopResumeRuleDryRun: return true default: return false diff --git a/internal/exporter/stop_resume_rule_scene.go b/internal/exporter/stop_resume_rule_scene.go new file mode 100644 index 00000000..0c4e2562 --- /dev/null +++ b/internal/exporter/stop_resume_rule_scene.go @@ -0,0 +1,98 @@ +package exporter + +import ( + "context" + "strconv" + + "gorm.io/gorm" + + "github.com/break/junhong_cmp_fiber/internal/query/stopresumerule" + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +// stopResumeRuleHeaders 停复机规则演练导出表头。 +var stopResumeRuleHeaders = []string{ + "卡ID", "ICCID", "运营商账户ID", "运营商账户", "绑定设备ID", "当前网络状态", "当前停机原因", "卡类别", + "满足实名条件", "判断结果", "原因", +} + +// StopResumeRuleDataSource 停复机规则演练明细导出数据源：按卡 ID 升序分页取卡并逐张按新规则判断，只读。 +// 仅超级管理员与平台账号可导出：受控入口已做角色门禁，这里再按任务内冻结的账号类型复核一次。 +type StopResumeRuleDataSource struct { + query *stopresumerule.Query +} + +// NewStopResumeRuleDataSource 创建停复机规则演练导出数据源。 +func NewStopResumeRuleDataSource(db *gorm.DB) *StopResumeRuleDataSource { + return &StopResumeRuleDataSource{query: stopresumerule.NewQuery(db)} +} + +// Scene 返回导出场景编码。 +func (s *StopResumeRuleDataSource) Scene() string { + return constants.ExportTaskSceneStopResumeRuleDryRun +} + +// Count 统计参与演练的卡数。 +func (s *StopResumeRuleDataSource) Count(ctx context.Context, params ExportParams) (int, error) { + filter, err := s.filter(params) + if err != nil { + return 0, err + } + total, err := s.query.Count(ctx, filter) + return int(total), err +} + +// Headers 返回导出表头。 +func (s *StopResumeRuleDataSource) Headers(ctx context.Context, params ExportParams) ([]string, error) { + if err := stopresumerule.EnsureAllowed(params.UserType); err != nil { + return nil, err + } + return stopResumeRuleHeaders, nil +} + +// Fetch 按 offset/limit 返回演练明细行。 +func (s *StopResumeRuleDataSource) Fetch(ctx context.Context, params ExportParams, offset, limit int) ([][]string, error) { + filter, err := s.filter(params) + if err != nil { + return nil, err + } + rows, err := s.query.Page(ctx, filter, offset, limit) + if err != nil { + return nil, err + } + result := make([][]string, 0, len(rows)) + for index := range rows { + item := stopresumerule.ToItem(&rows[index]) + network := "停机" + if item.NetworkStatus == constants.NetworkStatusOnline { + network = "开机" + } + device := "" + if item.DeviceID != 0 { + device = strconv.FormatUint(uint64(item.DeviceID), 10) + } + realname := "否" + if item.RealnameOK { + realname = "是" + } + result = append(result, []string{ + strconv.FormatUint(uint64(item.CardID), 10), item.ICCID, strconv.FormatUint(uint64(item.CarrierID), 10), + item.CarrierName, device, network, item.StopReasonName, item.CardCategory, realname, item.ResultName, item.ReasonName, + }) + } + return result, nil +} + +func (s *StopResumeRuleDataSource) filter(params ExportParams) (stopresumerule.Filter, error) { + if err := stopresumerule.EnsureAllowed(params.UserType); err != nil { + return stopresumerule.Filter{}, err + } + var carrierID *uint + if id, ok := filterUint(params.Filters, "carrier_id"); ok && id > 0 { + carrierID = &id + } + stopReason, _ := filterString(params.Filters, "stop_reason") + on, _ := filterString(params.Filters, "assume_control_on") + off, _ := filterString(params.Filters, "assume_control_off") + return stopresumerule.BuildFilter(carrierID, stopReason, on, off) +} diff --git a/internal/handler/admin/stop_resume_rule.go b/internal/handler/admin/stop_resume_rule.go new file mode 100644 index 00000000..462daacf --- /dev/null +++ b/internal/handler/admin/stop_resume_rule.go @@ -0,0 +1,81 @@ +package admin + +import ( + "github.com/go-playground/validator/v10" + "github.com/gofiber/fiber/v2" + + "github.com/break/junhong_cmp_fiber/internal/handler/validation" + "github.com/break/junhong_cmp_fiber/internal/model/dto" + "github.com/break/junhong_cmp_fiber/internal/query/stopresumerule" + exportTaskService "github.com/break/junhong_cmp_fiber/internal/service/export_task" + "github.com/break/junhong_cmp_fiber/pkg/constants" + "github.com/break/junhong_cmp_fiber/pkg/errors" + "github.com/break/junhong_cmp_fiber/pkg/middleware" + "github.com/break/junhong_cmp_fiber/pkg/response" +) + +// StopResumeRuleHandler 停复机规则演练 Handler：只判断不执行，仅超级管理员与平台账号可用。 +type StopResumeRuleHandler struct { + query *stopresumerule.Query + exportService *exportTaskService.Service + validator *validator.Validate +} + +// NewStopResumeRuleHandler 创建停复机规则演练 Handler。 +func NewStopResumeRuleHandler(query *stopresumerule.Query, exportService *exportTaskService.Service, validator *validator.Validate) *StopResumeRuleHandler { + return &StopResumeRuleHandler{query: query, exportService: exportService, validator: validator} +} + +// DryRun 执行停复机规则演练。 +// GET /api/admin/stop-resume-rule/dry-run +func (h *StopResumeRuleHandler) DryRun(c *fiber.Ctx) error { + var req dto.StopResumeRuleDryRunRequest + if err := c.QueryParser(&req); err != nil { + return errors.New(errors.CodeInvalidParam, "请求参数格式不正确") + } + if err := h.validator.Struct(&req); err != nil { + return errors.New(errors.CodeInvalidParam, validation.Message("停复机规则演练参数不正确", &req, err)) + } + result, err := h.query.DryRun(c.UserContext(), &req) + if err != nil { + return err + } + return response.Success(c, result) +} + +// Export 创建停复机规则演练明细异步导出任务。 +// POST /api/admin/stop-resume-rule/dry-run/export +func (h *StopResumeRuleHandler) Export(c *fiber.Ctx) error { + if err := stopresumerule.EnsureAllowed(middleware.GetUserTypeFromContext(c.UserContext())); err != nil { + return err + } + var req dto.StopResumeRuleDryRunExportRequest + if err := c.BodyParser(&req); err != nil { + return errors.New(errors.CodeInvalidParam, "请求参数格式不正确") + } + if err := h.validator.Struct(&req); err != nil { + return errors.New(errors.CodeInvalidParam, validation.Message("停复机规则演练导出参数不正确", &req, err)) + } + // 创建期校验筛选，避免任务执行时才因参数非法失败。 + if _, err := stopresumerule.BuildFilter(req.CarrierID, req.StopReason, req.AssumeControlOn, req.AssumeControlOff); err != nil { + return err + } + format := req.Format + if format == "" { + format = constants.ExportTaskFormatXLSX + } + filters := map[string]interface{}{ + "stop_reason": req.StopReason, "assume_control_on": req.AssumeControlOn, "assume_control_off": req.AssumeControlOff, + } + if req.CarrierID != nil { + filters["carrier_id"] = *req.CarrierID + } + result, err := h.exportService.CreateTask(c.UserContext(), &dto.CreateExportTaskRequest{ + Scene: constants.ExportTaskSceneStopResumeRuleDryRun, Format: format, + Query: map[string]interface{}{"filters": filters}, + }) + if err != nil { + return err + } + return response.Success(c, result) +} diff --git a/internal/infrastructure/audit/registry.go b/internal/infrastructure/audit/registry.go index fcdf677b..caa5a2f4 100644 --- a/internal/infrastructure/audit/registry.go +++ b/internal/infrastructure/audit/registry.go @@ -87,6 +87,17 @@ func NewRegistry() *Registry { iotCardDeactivated := iotCardAction(constants.AuditActionIotCardDeactivated, "停用 IoT 卡资产", constants.AuditActorAccount, constants.AuditSourceAdminAPI) iotCardDeactivated.Risk = constants.AuditRiskHigh iotCardPollingStatusUpdated := iotCardAction(constants.AuditActionIotCardPollingStatusUpdated, "更新 IoT 卡轮询开关", constants.AuditActorAccount, constants.AuditSourceAdminAPI) + iotCardTerminalPollingDisabled := iotCardAction(constants.AuditActionIotCardTerminalPollingDisabled, "上游终态自动关闭 IoT 卡轮询", constants.AuditActorSystemTask, constants.AuditSourceWorker) + iotCardTerminalPollingDisabled.AllowedActor, iotCardTerminalPollingDisabled.Source = "", "" + // 停复机可能由人工、Worker、定时任务、开放接口、C 端与回调触发，命中终态后均需留下关闭轮询的审计事实。 + iotCardTerminalPollingDisabled.AllowedOrigins = []ActionOrigin{ + {Actor: constants.AuditActorAccount, Source: constants.AuditSourceAdminAPI}, + {Actor: constants.AuditActorSystemTask, Source: constants.AuditSourceWorker}, + {Actor: constants.AuditActorScheduledJob, Source: constants.AuditSourceScheduler}, + {Actor: constants.AuditActorOpenAPI, Source: constants.AuditSourceOpenAPI}, + {Actor: constants.AuditActorPersonalCustomer, Source: constants.AuditSourcePersonalAPI}, + {Actor: constants.AuditActorExternalSystem, Source: constants.AuditSourceCallback}, + } iotCardPollingStatusBatchUpdated := iotCardBatchAction(constants.AuditActionIotCardPollingStatusBatchUpdated, "批量更新 IoT 卡轮询开关") iotCardBatchDeleted := ActionDefinition{ Code: constants.AuditActionIotCardBatchDeleted, Name: "批量删除 IoT 卡", @@ -527,6 +538,7 @@ func NewRegistry() *Registry { constants.AuditActionIotCardDeleted: iotCardDeleted, constants.AuditActionIotCardDeactivated: iotCardDeactivated, constants.AuditActionIotCardPollingStatusUpdated: iotCardPollingStatusUpdated, + constants.AuditActionIotCardTerminalPollingDisabled: iotCardTerminalPollingDisabled, constants.AuditActionIotCardPollingStatusBatchUpdated: iotCardPollingStatusBatchUpdated, constants.AuditActionIotCardBatchDeleted: iotCardBatchDeleted, constants.AuditActionIotCardAllocationBatch: iotCardAllocationBatch, diff --git a/internal/infrastructure/gatewaylimit/classifier.go b/internal/infrastructure/gatewaylimit/classifier.go index 086037cd..d44ea58d 100644 --- a/internal/infrastructure/gatewaylimit/classifier.go +++ b/internal/infrastructure/gatewaylimit/classifier.go @@ -24,6 +24,8 @@ type Classification struct { RuleID uint // 命中的规则 ID，未命中为 0 Keyword string // 命中的关键字 MaybeAccepted bool // 上游可能已受理 + // AutoDisablePolling 命中规则标记自动关闭轮询：停复机不重试，关闭该卡轮询。 + AutoDisablePolling bool } type compiledRule struct { @@ -81,7 +83,8 @@ func MatchRules(rules []compiledRule, op, message string) Classification { if !ruleMatches(rule, message) { continue } - return Classification{Category: rule.row.Category, RuleID: rule.row.ID, Keyword: rule.row.Keyword, MaybeAccepted: rule.row.MaybeAccepted} + return Classification{Category: rule.row.Category, RuleID: rule.row.ID, Keyword: rule.row.Keyword, + MaybeAccepted: rule.row.MaybeAccepted, AutoDisablePolling: rule.row.AutoDisablePolling && rule.row.Category == constants.GatewayClassTerminal} } return Classification{Category: constants.GatewayClassBusiness} } diff --git a/internal/infrastructure/gatewaylimit/runtime.go b/internal/infrastructure/gatewaylimit/runtime.go index dce45d62..774335e3 100644 --- a/internal/infrastructure/gatewaylimit/runtime.go +++ b/internal/infrastructure/gatewaylimit/runtime.go @@ -24,6 +24,8 @@ type Runtime struct { Limiter *Limiter Guard *Guard Recorder *StopResumeRecorder + // TerminalPolling 停复机命中"自动关闭轮询"终态规则时关闭该卡轮询。 + TerminalPolling *TerminalPollingDisabler redisClient *redis.Client logger *zap.Logger @@ -47,7 +49,9 @@ func NewRuntime(db *gorm.DB, redisClient *redis.Client, logger *zap.Logger) *Run limiter := NewLimiter(redisClient, groups, settings) guard := NewGuard(limiter, NewCarrierResolver(db, redisClient), classifier, redisClient, logger) recorder := NewStopResumeRecorder(postgres.NewCardStopResumeRecordStore(db), classifier, logger) - return &Runtime{Settings: settings, Groups: groups, Classifier: classifier, Limiter: limiter, Guard: guard, Recorder: recorder, redisClient: redisClient, logger: logger} + terminal := NewTerminalPollingDisabler(db, classifier, logger) + return &Runtime{Settings: settings, Groups: groups, Classifier: classifier, Limiter: limiter, Guard: guard, Recorder: recorder, + TerminalPolling: terminal, redisClient: redisClient, logger: logger} } // TrackDeferral 记录 Asynq 任务因限流被推迟；同一任务首次被限流起累计推迟超过阈值时写一次错误告警，任务继续重试。 diff --git a/internal/infrastructure/gatewaylimit/terminal_polling.go b/internal/infrastructure/gatewaylimit/terminal_polling.go new file mode 100644 index 00000000..6777b3b2 --- /dev/null +++ b/internal/infrastructure/gatewaylimit/terminal_polling.go @@ -0,0 +1,113 @@ +package gatewaylimit + +import ( + "context" + "strconv" + "strings" + "time" + + "go.uber.org/zap" + "gorm.io/gorm" + + "github.com/break/junhong_cmp_fiber/internal/gateway" + auditinfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/audit" + "github.com/break/junhong_cmp_fiber/internal/model" + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +// TerminalPollingDisabler 在停复机调用命中"自动关闭轮询"的终态规则时关闭该卡自身轮询： +// 不级联设备与设备下其他卡；同事务回写网关扩展状态并写审计。 +type TerminalPollingDisabler struct { + db *gorm.DB + classifier *Classifier + audit *auditinfra.Writer + logger *zap.Logger +} + +// NewTerminalPollingDisabler 创建终态关闭轮询处理器；db 或 classifier 为 nil 时不处理。 +func NewTerminalPollingDisabler(db *gorm.DB, classifier *Classifier, logger *zap.Logger) *TerminalPollingDisabler { + if logger == nil { + logger = zap.NewNop() + } + return &TerminalPollingDisabler{db: db, classifier: classifier, logger: logger} +} + +// SetAuditWriter 注入统一审计；未注入时只关闭轮询并记日志。 +func (d *TerminalPollingDisabler) SetAuditWriter(writer *auditinfra.Writer) { + if d != nil { + d.audit = writer + } +} + +// HandleCommandError 判断停复机上游失败是否命中"自动关闭轮询"规则；命中时关闭该卡轮询并返回 true， +// 调用方据此不再重试。action 为 constants.StopResumeActionStop / StopResumeActionStart。 +func (d *TerminalPollingDisabler) HandleCommandError(ctx context.Context, card *model.IotCard, action string, callErr error) bool { + if d == nil || d.db == nil || d.classifier == nil || card == nil || card.ID == 0 || callErr == nil { + return false + } + gwErr, ok := gateway.AsGatewayError(callErr) + if !ok || gwErr.Category != constants.GatewayErrorUpstream { + return false + } + op := constants.GatewayOpStop + if action == constants.StopResumeActionStart { + op = constants.GatewayOpStart + } + classification := d.classifier.Classify(ctx, op, gwErr.ProviderMessage) + if !classification.AutoDisablePolling { + return false + } + extend := terminalGatewayExtend(gwErr.ProviderMessage, card.GatewayExtend) + if err := d.disable(ctx, card, extend, classification, gwErr.ProviderMessage); err != nil { + d.logger.Error("上游终态自动关闭卡轮询失败", zap.Uint("card_id", card.ID), zap.String("iccid", card.ICCID), + zap.String("metric", "stop_resume.terminal_disable_polling.failed"), zap.Error(err)) + return true + } + d.logger.Warn("停复机命中终态规则，已关闭该卡轮询", + zap.Uint("card_id", card.ID), zap.String("iccid", card.ICCID), zap.String("action", action), + zap.Uint("rule_id", classification.RuleID), zap.String("keyword", classification.Keyword), + zap.String("gateway_extend", extend), zap.String("metric", "stop_resume.terminal_disable_polling")) + return true +} + +func (d *TerminalPollingDisabler) disable(ctx context.Context, card *model.IotCard, extend string, classification Classification, message string) error { + writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + defer cancel() + return d.db.WithContext(writeCtx).Transaction(func(tx *gorm.DB) error { + updates := map[string]any{"enable_polling": false, "gateway_extend": extend, "updated_at": time.Now()} + if err := tx.Model(&model.IotCard{}).Where("id = ?", card.ID).Updates(updates).Error; err != nil { + return err + } + if d.audit == nil { + return nil + } + id := strconv.FormatUint(uint64(card.ID), 10) + return d.audit.Append(ctx, tx, auditinfra.AppendInput{ + ActionCode: constants.AuditActionIotCardTerminalPollingDisabled, + Summary: "上游返回终态，自动关闭 IoT 卡轮询", + Result: constants.AuditResultSuccess, + Metadata: map[string]any{"rule_id": classification.RuleID, "keyword": classification.Keyword, + "provider_message": message}, + Resources: []auditinfra.ResourceInput{{ + Type: constants.AuditResourceIotCard, ID: &id, Key: auditinfra.IotCardResourceKey(card), DisplayName: card.ICCID, + Relation: constants.AuditResourceRelationPrimary, Role: constants.AuditResourceRoleIotCardTarget, + IdentitySnapshot: auditinfra.IotCardIdentitySnapshot(card), + BeforeData: map[string]any{"enable_polling": card.EnablePolling, "gateway_extend": card.GatewayExtend}, + AfterData: map[string]any{"enable_polling": false, "gateway_extend": extend}, + SubjectVisibility: constants.AuditSubjectResult, SubjectSummary: "IoT 卡已因上游终态关闭轮询", + }}, + }) + }) +} + +// terminalGatewayExtend 按上游返回推断网关扩展状态：已销户、风险停机；无法推断时保留原值。 +func terminalGatewayExtend(message, current string) string { + switch { + case strings.Contains(message, "销户"): + return constants.GatewayCardExtendCancelled + case strings.Contains(message, "风险"): + return constants.GatewayCardExtendRiskStop + default: + return current + } +} diff --git a/internal/model/dto/export_task_dto.go b/internal/model/dto/export_task_dto.go index 5d2112ca..73e785ee 100644 --- a/internal/model/dto/export_task_dto.go +++ b/internal/model/dto/export_task_dto.go @@ -4,7 +4,7 @@ import "time" // CreateExportTaskRequest 创建导出任务请求。 type CreateExportTaskRequest struct { - Scene string `json:"scene" validate:"required,oneof=device iot_card order package agent_wallet_transaction agent_recharge refund exchange commission_record package_traffic_alert expiring_asset operations_activation operations_renewal" required:"true" description:"导出场景 (device:设备, iot_card:IoT卡, order:订单, package:套餐, agent_wallet_transaction:代理主钱包流水, agent_recharge:代理充值, refund:退款, exchange:换货, commission_record:佣金明细, package_traffic_alert:套餐真流量达量预警, expiring_asset:临期资产, operations_activation:设备激活情况报表, operations_renewal:套餐续费情况报表)"` + Scene string `json:"scene" validate:"required,oneof=device iot_card order package agent_wallet_transaction agent_recharge refund exchange commission_record package_traffic_alert expiring_asset operations_activation operations_renewal stop_resume_rule_dry_run" required:"true" description:"导出场景 (device:设备, iot_card:IoT卡, order:订单, package:套餐, agent_wallet_transaction:代理主钱包流水, agent_recharge:代理充值, refund:退款, exchange:换货, commission_record:佣金明细, package_traffic_alert:套餐真流量达量预警, expiring_asset:临期资产, operations_activation:设备激活情况报表, operations_renewal:套餐续费情况报表, stop_resume_rule_dry_run:停复机规则演练明细)"` Format string `json:"format" validate:"required,oneof=xlsx csv" required:"true" description:"导出格式 (xlsx:Excel, csv:CSV)"` Query map[string]interface{} `json:"query,omitempty" description:"导出筛选参数(JSON对象，可选)；时间筛选固定使用 filters.start_time 与 filters.end_time，取值必须为带显式时区的 RFC3339 秒级时间"` } @@ -22,7 +22,7 @@ type CreateExportTaskResponse struct { type ListExportTaskRequest struct { Page int `json:"page" query:"page" validate:"omitempty,min=1" minimum:"1" description:"页码"` PageSize int `json:"page_size" query:"page_size" validate:"omitempty,min=1,max=100" minimum:"1" maximum:"100" description:"每页数量"` - Scene string `json:"scene" query:"scene" validate:"omitempty,oneof=device iot_card order package agent_wallet_transaction agent_recharge refund exchange commission_record package_traffic_alert expiring_asset operations_activation operations_renewal" description:"导出场景 (device:设备, iot_card:IoT卡, order:订单, package:套餐, agent_wallet_transaction:代理主钱包流水, agent_recharge:代理充值, refund:退款, exchange:换货, commission_record:佣金明细, package_traffic_alert:套餐真流量达量预警, expiring_asset:临期资产, operations_activation:设备激活情况报表, operations_renewal:套餐续费情况报表)"` + Scene string `json:"scene" query:"scene" validate:"omitempty,oneof=device iot_card order package agent_wallet_transaction agent_recharge refund exchange commission_record package_traffic_alert expiring_asset operations_activation operations_renewal stop_resume_rule_dry_run" description:"导出场景 (device:设备, iot_card:IoT卡, order:订单, package:套餐, agent_wallet_transaction:代理主钱包流水, agent_recharge:代理充值, refund:退款, exchange:换货, commission_record:佣金明细, package_traffic_alert:套餐真流量达量预警, expiring_asset:临期资产, operations_activation:设备激活情况报表, operations_renewal:套餐续费情况报表, stop_resume_rule_dry_run:停复机规则演练明细)"` Status *int `json:"status" query:"status" validate:"omitempty,min=1,max=5" minimum:"1" maximum:"5" description:"任务状态 (1:待处理, 2:处理中, 3:已完成, 4:已失败, 5:已取消)"` StartTime string `json:"start_time" query:"start_time" description:"创建时间起始（带时区的 RFC3339 秒级时间，闭区间含该时刻，如 2026-09-01T00:00:00+08:00）"` EndTime string `json:"end_time" query:"end_time" description:"创建时间结束（带时区的 RFC3339 秒级时间，闭区间含该时刻，如 2026-09-30T23:59:59+08:00）"` @@ -33,7 +33,7 @@ type ExportTaskItem struct { ID uint `json:"id" description:"任务ID"` TaskID uint `json:"task_id" description:"任务ID"` TaskNo string `json:"task_no" description:"任务编号"` - Scene string `json:"scene" description:"导出场景 (device:设备, iot_card:IoT卡, order:订单, package:套餐, agent_wallet_transaction:代理主钱包流水, agent_recharge:代理充值, refund:退款, exchange:换货, commission_record:佣金明细, package_traffic_alert:套餐真流量达量预警, operations_activation:设备激活情况报表, operations_renewal:套餐续费情况报表)"` + Scene string `json:"scene" description:"导出场景 (device:设备, iot_card:IoT卡, order:订单, package:套餐, agent_wallet_transaction:代理主钱包流水, agent_recharge:代理充值, refund:退款, exchange:换货, commission_record:佣金明细, package_traffic_alert:套餐真流量达量预警, operations_activation:设备激活情况报表, operations_renewal:套餐续费情况报表, stop_resume_rule_dry_run:停复机规则演练明细)"` Format string `json:"format" description:"导出格式 (xlsx:Excel, csv:CSV)"` Status int `json:"status" description:"任务状态 (1:待处理, 2:处理中, 3:已完成, 4:已失败, 5:已取消)"` StatusName string `json:"status_name" description:"任务状态名称（中文）"` diff --git a/internal/model/dto/gateway_rate_limit_dto.go b/internal/model/dto/gateway_rate_limit_dto.go index 3120d968..062ad2c0 100644 --- a/internal/model/dto/gateway_rate_limit_dto.go +++ b/internal/model/dto/gateway_rate_limit_dto.go @@ -74,15 +74,16 @@ type GatewayErrorRuleListRequest struct { // SaveGatewayErrorRuleRequest 新建或修改上游错误分类规则请求 type SaveGatewayErrorRuleRequest struct { - ID uint `json:"-" description:"规则ID"` - Keyword string `json:"keyword" validate:"required,max=200" required:"true" maxLength:"200" description:"匹配关键字或正则"` - MatchType string `json:"match_type" validate:"required,oneof=contains prefix regex" required:"true" enum:"contains,prefix,regex" description:"匹配方式 (contains:包含, prefix:前缀, regex:正则)"` - Category string `json:"category" validate:"required,oneof=rate_limited_upstream terminal already_target transient business account_config" required:"true" enum:"rate_limited_upstream,terminal,already_target,transient,business,account_config" description:"类别 (rate_limited_upstream:上游频率限制, terminal:终态不可操作, already_target:已是目标状态, transient:临时故障, business:业务错误, account_config:账户配置错误)"` - Operations []GatewayOperation `json:"operations" validate:"omitempty,dive,oneof=query_flow query_status query_realname stop start" description:"适用操作，为空表示全部"` - Priority int `json:"priority" validate:"min=1,max=10000" minimum:"1" maximum:"10000" description:"匹配优先级，越小越先匹配"` - MaybeAccepted bool `json:"maybe_accepted" description:"上游可能已受理（如返回含调用成功、操作进行中、请求被合并）"` - Status *int `json:"status" validate:"omitempty,oneof=0 1" enum:"0,1" description:"状态 (0:停用, 1:启用)，新建默认启用"` - Remark string `json:"remark" validate:"omitempty,max=500" maxLength:"500" description:"备注"` + ID uint `json:"-" description:"规则ID"` + Keyword string `json:"keyword" validate:"required,max=200" required:"true" maxLength:"200" description:"匹配关键字或正则"` + MatchType string `json:"match_type" validate:"required,oneof=contains prefix regex" required:"true" enum:"contains,prefix,regex" description:"匹配方式 (contains:包含, prefix:前缀, regex:正则)"` + Category string `json:"category" validate:"required,oneof=rate_limited_upstream terminal already_target transient business account_config" required:"true" enum:"rate_limited_upstream,terminal,already_target,transient,business,account_config" description:"类别 (rate_limited_upstream:上游频率限制, terminal:终态不可操作, already_target:已是目标状态, transient:临时故障, business:业务错误, account_config:账户配置错误)"` + Operations []GatewayOperation `json:"operations" validate:"omitempty,dive,oneof=query_flow query_status query_realname stop start" description:"适用操作，为空表示全部"` + Priority int `json:"priority" validate:"min=1,max=10000" minimum:"1" maximum:"10000" description:"匹配优先级，越小越先匹配"` + MaybeAccepted bool `json:"maybe_accepted" description:"上游可能已受理（如返回含调用成功、操作进行中、请求被合并）"` + AutoDisablePolling *bool `json:"auto_disable_polling" description:"命中后自动关闭该卡轮询（停复机不重试、回写网关扩展状态），仅类别为 terminal 时可勾选；更新时不传保留原值"` + Status *int `json:"status" validate:"omitempty,oneof=0 1" enum:"0,1" description:"状态 (0:停用, 1:启用)，新建默认启用"` + Remark string `json:"remark" validate:"omitempty,max=500" maxLength:"500" description:"备注"` } // UpdateGatewayErrorRuleStatusRequest 启停上游错误分类规则请求 @@ -93,18 +94,19 @@ type UpdateGatewayErrorRuleStatusRequest struct { // GatewayErrorRuleResponse 上游错误分类规则响应 type GatewayErrorRuleResponse struct { - ID uint `json:"id" description:"规则ID"` - Keyword string `json:"keyword" description:"匹配关键字或正则"` - MatchType string `json:"match_type" description:"匹配方式 (contains:包含, prefix:前缀, regex:正则)"` - Category string `json:"category" description:"类别"` - CategoryName string `json:"category_name" description:"类别名称"` - Operations []string `json:"operations" description:"适用操作，为空表示全部"` - Priority int `json:"priority" description:"匹配优先级"` - MaybeAccepted bool `json:"maybe_accepted" description:"上游可能已受理"` - Status int `json:"status" description:"状态 (0:停用, 1:启用)"` - Remark string `json:"remark" description:"备注"` - CreatedAt time.Time `json:"created_at" description:"创建时间"` - UpdatedAt time.Time `json:"updated_at" description:"更新时间"` + ID uint `json:"id" description:"规则ID"` + Keyword string `json:"keyword" description:"匹配关键字或正则"` + MatchType string `json:"match_type" description:"匹配方式 (contains:包含, prefix:前缀, regex:正则)"` + Category string `json:"category" description:"类别"` + CategoryName string `json:"category_name" description:"类别名称"` + Operations []string `json:"operations" description:"适用操作，为空表示全部"` + Priority int `json:"priority" description:"匹配优先级"` + MaybeAccepted bool `json:"maybe_accepted" description:"上游可能已受理"` + AutoDisablePolling bool `json:"auto_disable_polling" description:"命中后自动关闭该卡轮询"` + Status int `json:"status" description:"状态 (0:停用, 1:启用)"` + Remark string `json:"remark" description:"备注"` + CreatedAt time.Time `json:"created_at" description:"创建时间"` + UpdatedAt time.Time `json:"updated_at" description:"更新时间"` } // GatewayErrorRulePageResult 上游错误分类规则分页结果 @@ -124,12 +126,13 @@ type TestGatewayErrorRuleRequest struct { // TestGatewayErrorRuleResponse 测试上游错误分类响应 type TestGatewayErrorRuleResponse struct { - Category string `json:"category" description:"命中类别"` - CategoryName string `json:"category_name" description:"命中类别名称"` - RuleID uint `json:"rule_id" description:"命中规则ID，未命中或命中草稿为 0"` - Keyword string `json:"keyword" description:"命中关键字，未命中为空"` - MaybeAccepted bool `json:"maybe_accepted" description:"上游可能已受理"` - Matched bool `json:"matched" description:"是否命中规则；未命中时按业务错误处理"` + Category string `json:"category" description:"命中类别"` + CategoryName string `json:"category_name" description:"命中类别名称"` + RuleID uint `json:"rule_id" description:"命中规则ID，未命中或命中草稿为 0"` + Keyword string `json:"keyword" description:"命中关键字，未命中为空"` + MaybeAccepted bool `json:"maybe_accepted" description:"上游可能已受理"` + AutoDisablePolling bool `json:"auto_disable_polling" description:"命中后自动关闭该卡轮询"` + Matched bool `json:"matched" description:"是否命中规则；未命中时按业务错误处理"` } // StopResumeRecordListRequest 停复机记录列表请求 diff --git a/internal/model/dto/stop_resume_rule_dto.go b/internal/model/dto/stop_resume_rule_dto.go new file mode 100644 index 00000000..741c209e --- /dev/null +++ b/internal/model/dto/stop_resume_rule_dto.go @@ -0,0 +1,70 @@ +package dto + +// StopResumeRuleDryRunRequest 停复机规则演练请求。 +type StopResumeRuleDryRunRequest struct { + CarrierID *uint `json:"carrier_id" query:"carrier_id" validate:"omitempty,min=1" minimum:"1" description:"运营商账户ID，不传为全部"` + StopReason string `json:"stop_reason" query:"stop_reason" validate:"omitempty,max=50" maxLength:"50" description:"停机原因筛选：传停机原因编码（如 not_realname、carrier_stopped、manual），传 none 表示停机原因为空，不传为全部"` + AssumeControlOn string `json:"assume_control_on" query:"assume_control_on" validate:"omitempty,max=2000" maxLength:"2000" description:"假设开启实名管控的运营商账户ID，逗号分隔；只用于预演，不改实际配置"` + AssumeControlOff string `json:"assume_control_off" query:"assume_control_off" validate:"omitempty,max=2000" maxLength:"2000" description:"假设关闭实名管控的运营商账户ID，逗号分隔；只用于预演，不改实际配置"` + Result string `json:"result" query:"result" validate:"omitempty,oneof=will_resume will_stop no_action" enum:"will_resume,will_stop,no_action" description:"明细按判断结果筛选 (will_resume:将复机, will_stop:将停机, no_action:不动作)，汇总不受影响"` + Page int `json:"page" query:"page" validate:"omitempty,min=1" minimum:"1" description:"明细页码，默认1"` + PageSize int `json:"page_size" query:"page_size" validate:"omitempty,min=1,max=500" minimum:"1" maximum:"500" description:"明细每页数量，默认20，最大500"` +} + +// StopResumeRuleDryRunExportRequest 停复机规则演练明细导出请求。 +type StopResumeRuleDryRunExportRequest struct { + CarrierID *uint `json:"carrier_id" validate:"omitempty,min=1" minimum:"1" description:"运营商账户ID，不传为全部"` + StopReason string `json:"stop_reason" validate:"omitempty,max=50" maxLength:"50" description:"停机原因筛选，none 表示停机原因为空"` + AssumeControlOn string `json:"assume_control_on" validate:"omitempty,max=2000" maxLength:"2000" description:"假设开启实名管控的运营商账户ID，逗号分隔"` + AssumeControlOff string `json:"assume_control_off" validate:"omitempty,max=2000" maxLength:"2000" description:"假设关闭实名管控的运营商账户ID，逗号分隔"` + Format string `json:"format" validate:"omitempty,oneof=xlsx csv" enum:"xlsx,csv" description:"导出格式 (xlsx:Excel, csv:CSV)，默认 xlsx"` +} + +// StopResumeRuleDryRunSummaryItem 演练汇总项：运营商账户 × 当前停机原因 × 判断结果 × 原因。 +type StopResumeRuleDryRunSummaryItem struct { + CarrierID uint `json:"carrier_id" description:"运营商账户ID"` + CarrierName string `json:"carrier_name" description:"运营商账户名称"` + StopReason string `json:"stop_reason" description:"当前停机原因，空表示无"` + StopReasonName string `json:"stop_reason_name" description:"当前停机原因名称"` + Result string `json:"result" description:"判断结果 (will_resume:将复机, will_stop:将停机, no_action:不动作)"` + ResultName string `json:"result_name" description:"判断结果名称"` + Reason string `json:"reason" description:"将停机时为目标停机原因，不动作时为不动作原因 (polling_suspended:暂停轮询, conditions_met:满足开机条件, no_package:无套餐, traffic_exhausted:流量已用完, not_realname:未实名, threshold_locked:持通道阈值锁, carrier_missing:所属通道不存在)"` + ReasonName string `json:"reason_name" description:"原因名称"` + Count int64 `json:"count" description:"卡数"` +} + +// StopResumeRuleDryRunTotals 演练总计。 +type StopResumeRuleDryRunTotals struct { + Total int64 `json:"total" description:"参与演练的开机与停机卡数"` + WillResume int64 `json:"will_resume" description:"将复机卡数"` + WillStop int64 `json:"will_stop" description:"将停机卡数"` + NoAction int64 `json:"no_action" description:"不动作卡数"` +} + +// StopResumeRuleDryRunItem 演练明细。 +type StopResumeRuleDryRunItem struct { + CardID uint `json:"card_id" description:"卡ID"` + ICCID string `json:"iccid" description:"ICCID"` + CarrierID uint `json:"carrier_id" description:"运营商账户ID"` + CarrierName string `json:"carrier_name" description:"运营商账户名称"` + DeviceID uint `json:"device_id" description:"有效绑定设备ID，独立卡为 0"` + NetworkStatus int `json:"network_status" description:"当前网络状态 (0:停机, 1:开机)"` + StopReason string `json:"stop_reason" description:"当前停机原因"` + StopReasonName string `json:"stop_reason_name" description:"当前停机原因名称"` + CardCategory string `json:"card_category" description:"卡类别"` + RealnameOK bool `json:"realname_ok" description:"按（假设的）通道实名管控是否满足实名条件"` + Result string `json:"result" description:"判断结果"` + ResultName string `json:"result_name" description:"判断结果名称"` + Reason string `json:"reason" description:"将停机时为目标停机原因，不动作时为不动作原因"` + ReasonName string `json:"reason_name" description:"原因名称"` +} + +// StopResumeRuleDryRunResponse 停复机规则演练响应。 +type StopResumeRuleDryRunResponse struct { + Totals StopResumeRuleDryRunTotals `json:"totals" description:"总计"` + Summary []StopResumeRuleDryRunSummaryItem `json:"summary" description:"按运营商账户 × 当前停机原因 × 判断结果 × 原因汇总"` + Items []StopResumeRuleDryRunItem `json:"items" description:"明细（按卡ID升序分页）"` + Total int64 `json:"total" description:"明细总数（按 result 筛选后）"` + Page int `json:"page" description:"当前页"` + PageSize int `json:"size" description:"每页数量"` +} diff --git a/internal/model/gateway_rate_limit.go b/internal/model/gateway_rate_limit.go index 2e2cb559..53304682 100644 --- a/internal/model/gateway_rate_limit.go +++ b/internal/model/gateway_rate_limit.go @@ -37,8 +37,10 @@ type GatewayErrorClassification struct { Operations datatypes.JSON `gorm:"column:operations;type:jsonb;not null;comment:适用操作，空数组表示全部" json:"operations"` Priority int `gorm:"column:priority;not null;default:100;comment:匹配优先级" json:"priority"` MaybeAccepted bool `gorm:"column:maybe_accepted;not null;default:false;comment:上游可能已受理" json:"maybe_accepted"` - Status int `gorm:"column:status;type:smallint;not null;default:1;comment:状态 0-停用 1-启用" json:"status"` - Remark string `gorm:"column:remark;type:varchar(500);not null;default:'';comment:备注" json:"remark"` + // AutoDisablePolling 命中后自动关闭该卡轮询，仅终态不可操作类别可置位。 + AutoDisablePolling bool `gorm:"column:auto_disable_polling;not null;default:false;comment:命中后自动关闭该卡轮询" json:"auto_disable_polling"` + Status int `gorm:"column:status;type:smallint;not null;default:1;comment:状态 0-停用 1-启用" json:"status"` + Remark string `gorm:"column:remark;type:varchar(500);not null;default:'';comment:备注" json:"remark"` } // TableName 返回错误分类规则表名。 diff --git a/internal/query/stopresumerule/dry_run.go b/internal/query/stopresumerule/dry_run.go new file mode 100644 index 00000000..06900d36 --- /dev/null +++ b/internal/query/stopresumerule/dry_run.go @@ -0,0 +1,181 @@ +package stopresumerule + +import ( + "context" + "sort" + "strconv" + "strings" + + "github.com/break/junhong_cmp_fiber/internal/domain/stopresume" + "github.com/break/junhong_cmp_fiber/internal/model/dto" + "github.com/break/junhong_cmp_fiber/pkg/constants" + "github.com/break/junhong_cmp_fiber/pkg/errors" + "github.com/break/junhong_cmp_fiber/pkg/middleware" +) + +// 明细分页默认值与上限。 +const ( + defaultPageSize = 20 + maxPageSize = 500 +) + +// EnsureAllowed 校验演练与导出权限：仅超级管理员与平台账号。 +func EnsureAllowed(userType int) error { + if userType == constants.UserTypeSuperAdmin || userType == constants.UserTypePlatform { + return nil + } + return errors.New(errors.CodeForbidden, constants.PlatformManagementForbiddenMessage) +} + +// BuildFilter 把请求参数转换为演练筛选；假设开启与关闭的账户不得重叠。 +func BuildFilter(carrierID *uint, stopReason, assumeOn, assumeOff string) (Filter, error) { + on, err := parseIDs(assumeOn, "assume_control_on") + if err != nil { + return Filter{}, err + } + off, err := parseIDs(assumeOff, "assume_control_off") + if err != nil { + return Filter{}, err + } + control := make(map[uint]bool, len(on)+len(off)) + for _, id := range on { + control[id] = true + } + for _, id := range off { + if control[id] { + return Filter{}, errors.New(errors.CodeInvalidParam, "同一运营商账户不能同时假设开启和关闭实名管控") + } + control[id] = false + } + return Filter{CarrierID: carrierID, StopReason: strings.TrimSpace(stopReason), RealnameControl: control}, nil +} + +func parseIDs(raw, field string) ([]uint, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return nil, nil + } + parts := strings.Split(raw, ",") + ids := make([]uint, 0, len(parts)) + for _, part := range parts { + part = strings.TrimSpace(part) + if part == "" { + continue + } + id, err := strconv.ParseUint(part, 10, 64) + if err != nil || id == 0 { + return nil, errors.New(errors.CodeInvalidParam, field+" 只能是逗号分隔的运营商账户ID") + } + ids = append(ids, uint(id)) + } + return ids, nil +} + +type summaryKey struct { + carrierID uint + stopReason string + result string + reason string +} + +// DryRun 执行演练：全量分批判断，返回汇总与按 result 筛选后的分页明细。只读。 +func (q *Query) DryRun(ctx context.Context, req *dto.StopResumeRuleDryRunRequest) (*dto.StopResumeRuleDryRunResponse, error) { + if err := EnsureAllowed(middleware.GetUserTypeFromContext(ctx)); err != nil { + return nil, err + } + filter, err := BuildFilter(req.CarrierID, req.StopReason, req.AssumeControlOn, req.AssumeControlOff) + if err != nil { + return nil, err + } + page, size := req.Page, req.PageSize + if page < 1 { + page = 1 + } + if size < 1 { + size = defaultPageSize + } + if size > maxPageSize { + size = maxPageSize + } + offset := (page - 1) * size + resp := &dto.StopResumeRuleDryRunResponse{Page: page, PageSize: size, Items: []dto.StopResumeRuleDryRunItem{}} + counts := map[summaryKey]int64{} + names := map[uint]string{} + var matched int64 + err = q.Scan(ctx, filter, func(rows []Row) error { + for index := range rows { + row := &rows[index] + resp.Totals.Total++ + switch row.Result { + case stopresume.ResultWillResume: + resp.Totals.WillResume++ + case stopresume.ResultWillStop: + resp.Totals.WillStop++ + default: + resp.Totals.NoAction++ + } + counts[summaryKey{row.CarrierID, row.StopReason, row.Result, row.Reason}]++ + names[row.CarrierID] = row.CarrierName + if req.Result != "" && row.Result != req.Result { + continue + } + if matched >= int64(offset) && len(resp.Items) < size { + resp.Items = append(resp.Items, ToItem(row)) + } + matched++ + } + return nil + }) + if err != nil { + return nil, err + } + resp.Total = matched + resp.Summary = buildSummary(counts, names) + return resp, nil +} + +func buildSummary(counts map[summaryKey]int64, names map[uint]string) []dto.StopResumeRuleDryRunSummaryItem { + items := make([]dto.StopResumeRuleDryRunSummaryItem, 0, len(counts)) + for key, count := range counts { + items = append(items, dto.StopResumeRuleDryRunSummaryItem{ + CarrierID: key.carrierID, CarrierName: names[key.carrierID], + StopReason: key.stopReason, StopReasonName: stopReasonName(key.stopReason), + Result: key.result, ResultName: stopresume.ResultName(key.result), + Reason: key.reason, ReasonName: stopresume.ReasonName(key.reason), Count: count, + }) + } + sort.Slice(items, func(i, j int) bool { + a, b := items[i], items[j] + if a.CarrierID != b.CarrierID { + return a.CarrierID < b.CarrierID + } + if a.StopReason != b.StopReason { + return a.StopReason < b.StopReason + } + if a.Result != b.Result { + return a.Result < b.Result + } + return a.Reason < b.Reason + }) + return items +} + +// ToItem 把演练行转换为接口明细。 +func ToItem(row *Row) dto.StopResumeRuleDryRunItem { + return dto.StopResumeRuleDryRunItem{ + CardID: row.CardID, ICCID: row.ICCID, CarrierID: row.CarrierID, CarrierName: row.CarrierName, + DeviceID: row.DeviceID, NetworkStatus: row.NetworkStatus, + StopReason: row.StopReason, StopReasonName: stopReasonName(row.StopReason), + CardCategory: row.CardCategory, RealnameOK: row.RealnameOK, + Result: row.Result, ResultName: stopresume.ResultName(row.Result), + Reason: row.Reason, ReasonName: stopresume.ReasonName(row.Reason), + } +} + +// stopReasonName 返回当前停机原因名称；空停因显示为"无"。 +func stopReasonName(reason string) string { + if strings.TrimSpace(reason) == "" { + return "无" + } + return constants.AgentOpenAPIStopReasonName(reason) +} diff --git a/internal/query/stopresumerule/query.go b/internal/query/stopresumerule/query.go new file mode 100644 index 00000000..6b23c1e4 --- /dev/null +++ b/internal/query/stopresumerule/query.go @@ -0,0 +1,329 @@ +// Package stopresumerule 提供停复机规则演练的只读查询：按批读取卡与资产当前事实，按新规则判断目标动作。 +// +// 演练不调用网关、不写任何卡、指令或调度状态；始终按新规则判断，不受停复机总开关影响； +// 可传入假设的通道实名管控配置预演，实际配置不变。 +package stopresumerule + +import ( + "context" + "time" + + "gorm.io/gorm" + + carrierthresholddomain "github.com/break/junhong_cmp_fiber/internal/domain/carrierthreshold" + "github.com/break/junhong_cmp_fiber/internal/domain/stopresume" + "github.com/break/junhong_cmp_fiber/internal/model" + "github.com/break/junhong_cmp_fiber/pkg/constants" + "github.com/break/junhong_cmp_fiber/pkg/errors" +) + +// batchSize 是演练每批读取的卡数。 +const batchSize = 1000 + +// StopReasonEmpty 是筛选"停机原因为空"时使用的取值。 +const StopReasonEmpty = "none" + +// Filter 是演练筛选条件。 +type Filter struct { + CarrierID *uint + StopReason string // 为空不限；StopReasonEmpty 表示停机原因为空 + // RealnameControl 假设的通道实名管控：键为运营商账户 ID，值为是否开启；未列出的账户按实际配置。 + RealnameControl map[uint]bool +} + +// Row 是一张卡的演练结果。 +type Row struct { + CardID uint + ICCID string + CarrierID uint + CarrierName string + DeviceID uint // 无有效绑定为 0 + NetworkStatus int + StopReason string + CardCategory string + RealnameOK bool + stopresume.Decision +} + +// Query 停复机规则演练查询。 +type Query struct { + db *gorm.DB + now func() time.Time +} + +// NewQuery 创建演练查询。 +func NewQuery(db *gorm.DB) *Query { + return &Query{db: db, now: time.Now} +} + +// Scan 按卡 ID 升序分批演练全部符合筛选的开机与停机卡，每批回调一次。 +func (q *Query) Scan(ctx context.Context, filter Filter, fn func([]Row) error) error { + carriers, err := q.loadCarriers(ctx) + if err != nil { + return err + } + var lastID uint + for { + if err := ctx.Err(); err != nil { + return err + } + var cards []model.IotCard + if err := q.cardQuery(ctx, filter).Where("id > ?", lastID).Order("id ASC").Limit(batchSize).Find(&cards).Error; err != nil { + return errors.Wrap(errors.CodeDatabaseError, err, "查询演练卡失败") + } + if len(cards) == 0 { + return nil + } + lastID = cards[len(cards)-1].ID + rows, err := q.evaluate(ctx, cards, carriers, filter) + if err != nil { + return err + } + if err := fn(rows); err != nil { + return err + } + } +} + +// Count 统计符合筛选的卡数（导出行数）。 +func (q *Query) Count(ctx context.Context, filter Filter) (int64, error) { + var total int64 + if err := q.cardQuery(ctx, filter).Count(&total).Error; err != nil { + return 0, errors.Wrap(errors.CodeDatabaseError, err, "统计演练卡数失败") + } + return total, nil +} + +// Page 按卡 ID 升序取 offset/limit 范围内的卡并演练，供导出分片使用。 +func (q *Query) Page(ctx context.Context, filter Filter, offset, limit int) ([]Row, error) { + carriers, err := q.loadCarriers(ctx) + if err != nil { + return nil, err + } + var cards []model.IotCard + if err := q.cardQuery(ctx, filter).Order("id ASC").Offset(offset).Limit(limit).Find(&cards).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询演练卡失败") + } + return q.evaluate(ctx, cards, carriers, filter) +} + +func (q *Query) cardQuery(ctx context.Context, filter Filter) *gorm.DB { + query := q.db.WithContext(ctx).Model(&model.IotCard{}). + Select("id", "iccid", "carrier_id", "network_status", "stop_reason", "enable_polling", "real_name_status", + "card_category", "is_standalone"). + Where("network_status IN ?", []int{constants.NetworkStatusOffline, constants.NetworkStatusOnline}) + if filter.CarrierID != nil { + query = query.Where("carrier_id = ?", *filter.CarrierID) + } + switch filter.StopReason { + case "": + case StopReasonEmpty: + query = query.Where("COALESCE(stop_reason, '') = ''") + default: + query = query.Where("stop_reason = ?", filter.StopReason) + } + return query +} + +type carrierFacts struct { + name string + realnameControl bool + resetDay int +} + +func (q *Query) loadCarriers(ctx context.Context) (map[uint]carrierFacts, error) { + var carriers []model.Carrier + if err := q.db.WithContext(ctx).Select("id", "carrier_name", "realname_control_enabled", "data_reset_day"). + Find(&carriers).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询运营商账户失败") + } + result := make(map[uint]carrierFacts, len(carriers)) + for _, carrier := range carriers { + result[carrier.ID] = carrierFacts{ + name: carrier.CarrierName, realnameControl: carrier.RealnameControlEnabled == 1, resetDay: carrier.DataResetDay, + } + } + return result, nil +} + +// evaluate 批量加载一批卡的绑定、设备轮询开关、套餐与通道阈值锁，再逐卡判断。 +func (q *Query) evaluate(ctx context.Context, cards []model.IotCard, carriers map[uint]carrierFacts, filter Filter) ([]Row, error) { + if len(cards) == 0 { + return nil, nil + } + cardIDs := make([]uint, 0, len(cards)) + for _, card := range cards { + cardIDs = append(cardIDs, card.ID) + } + deviceOf, err := q.activeBindings(ctx, cards) + if err != nil { + return nil, err + } + deviceIDs := uniqueValues(deviceOf) + devicePolling, err := q.devicePolling(ctx, deviceIDs) + if err != nil { + return nil, err + } + cardPackages, err := q.packages(ctx, "iot_card_id", cardIDs) + if err != nil { + return nil, err + } + devicePackages, err := q.packages(ctx, "device_id", deviceIDs) + if err != nil { + return nil, err + } + locked, err := q.lockedCards(ctx, cardIDs, carriers) + if err != nil { + return nil, err + } + rows := make([]Row, 0, len(cards)) + for _, card := range cards { + deviceID := deviceOf[card.ID] + facts := packageFactsOf(cardPackages[card.ID]) + suspended := !card.EnablePolling + if deviceID != 0 { + facts = packageFactsOf(devicePackages[deviceID]) + if enabled, ok := devicePolling[deviceID]; ok && !enabled { + suspended = true + } + } + carrier, known := carriers[card.CarrierID] + control := carrier.realnameControl + if assumed, ok := filter.RealnameControl[card.CarrierID]; ok { + control = assumed + } + verified := card.RealNameStatus == constants.RealNameStatusVerified + decision := stopresume.Decide(stopresume.Facts{ + NetworkStatus: card.NetworkStatus, PollingSuspended: suspended, + HasValidPackage: facts.hasValid, TrafficExhausted: facts.exhausted, + RealnameVerified: verified, CarrierKnown: known, RealnameControl: control, + ThresholdLocked: locked[card.ID], CurrentStopReason: card.StopReason, + }) + rows = append(rows, Row{ + CardID: card.ID, ICCID: card.ICCID, CarrierID: card.CarrierID, CarrierName: carrier.name, DeviceID: deviceID, + NetworkStatus: card.NetworkStatus, StopReason: card.StopReason, CardCategory: card.CardCategory, + RealnameOK: !control || verified, Decision: decision, + }) + } + return rows, nil +} + +// activeBindings 返回非独立卡的有效绑定设备；与停复机评估一致，独立卡不查绑定。 +func (q *Query) activeBindings(ctx context.Context, cards []model.IotCard) (map[uint]uint, error) { + bound := make([]uint, 0, len(cards)) + for _, card := range cards { + if !card.IsStandalone { + bound = append(bound, card.ID) + } + } + result := make(map[uint]uint, len(bound)) + if len(bound) == 0 { + return result, nil + } + var bindings []model.DeviceSimBinding + if err := q.db.WithContext(ctx).Select("iot_card_id", "device_id"). + Where("iot_card_id IN ? AND bind_status = ?", bound, constants.BindStatusBound). + Find(&bindings).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询卡绑定设备失败") + } + for _, binding := range bindings { + result[binding.IotCardID] = binding.DeviceID + } + return result, nil +} + +func (q *Query) devicePolling(ctx context.Context, deviceIDs []uint) (map[uint]bool, error) { + result := make(map[uint]bool, len(deviceIDs)) + if len(deviceIDs) == 0 { + return result, nil + } + var devices []model.Device + if err := q.db.WithContext(ctx).Select("id", "enable_polling").Where("id IN ?", deviceIDs).Find(&devices).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询设备轮询状态失败") + } + for _, device := range devices { + result[device.ID] = device.EnablePolling + } + return result, nil +} + +// packages 读取载体生效中与已用完的套餐，按载体分组。 +func (q *Query) packages(ctx context.Context, column string, ids []uint) (map[uint][]model.PackageUsage, error) { + result := make(map[uint][]model.PackageUsage, len(ids)) + if len(ids) == 0 { + return result, nil + } + var usages []model.PackageUsage + if err := q.db.WithContext(ctx).Where(column+" IN ?", ids). + Where("status IN ?", []int{constants.PackageUsageStatusActive, constants.PackageUsageStatusDepleted}). + Find(&usages).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询套餐使用记录失败") + } + for _, usage := range usages { + key := usage.IotCardID + if column == "device_id" { + key = usage.DeviceID + } + result[key] = append(result[key], usage) + } + return result, nil +} + +type packageFacts struct { + hasValid bool + exhausted bool +} + +// packageFactsOf 与评估主流程同口径：有生效中套餐即有套餐；生效或已用完的套餐全部达到上限才算流量用完。 +func packageFactsOf(usages []model.PackageUsage) packageFacts { + facts := packageFacts{} + if len(usages) == 0 { + return facts + } + facts.exhausted = true + for index := range usages { + if usages[index].Status == constants.PackageUsageStatusActive { + facts.hasValid = true + } + if !usages[index].IsTrafficDepleted() { + facts.exhausted = false + } + } + return facts +} + +// lockedCards 返回持本周期通道阈值锁的卡；与停复机评估同口径，账户不存在或重置日非法时按仍生效处理。 +func (q *Query) lockedCards(ctx context.Context, cardIDs []uint, carriers map[uint]carrierFacts) (map[uint]bool, error) { + result := make(map[uint]bool) + var locks []model.CarrierTrafficThresholdLock + if err := q.db.WithContext(ctx).Select("card_id", "carrier_id", "period_start"). + Where("card_id IN ? AND status = ?", cardIDs, carrierthresholddomain.LockStatusLocked). + Find(&locks).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询通道阈值锁失败") + } + now := q.now() + for _, lock := range locks { + carrier, ok := carriers[lock.CarrierID] + if !ok { + result[lock.CardID] = true + continue + } + periodStart, err := carrierthresholddomain.PeriodStart(now, carrier.resetDay) + if err != nil || lock.PeriodStart.Equal(periodStart) { + result[lock.CardID] = true + } + } + return result, nil +} + +func uniqueValues(m map[uint]uint) []uint { + seen := make(map[uint]bool, len(m)) + result := make([]uint, 0, len(m)) + for _, value := range m { + if value != 0 && !seen[value] { + seen[value] = true + result = append(result, value) + } + } + return result +} diff --git a/internal/routes/admin.go b/internal/routes/admin.go index 9c999627..765b8721 100644 --- a/internal/routes/admin.go +++ b/internal/routes/admin.go @@ -112,6 +112,9 @@ func RegisterAdminRoutes(router fiber.Router, handlers *bootstrap.Handlers, midd if handlers.AssetAutoRenewal != nil { registerAssetAutoRenewalRoutes(authGroup, handlers.AssetAutoRenewal, doc, basePath) } + if handlers.StopResumeRule != nil { + registerStopResumeRuleRoutes(authGroup, handlers.StopResumeRule, doc, basePath) + } if handlers.OperationsReport != nil { registerOperationsReportRoutes(authGroup, handlers.OperationsReport, doc, basePath) } diff --git a/internal/routes/stop_resume_rule.go b/internal/routes/stop_resume_rule.go new file mode 100644 index 00000000..3218e581 --- /dev/null +++ b/internal/routes/stop_resume_rule.go @@ -0,0 +1,46 @@ +package routes + +import ( + "github.com/gofiber/fiber/v2" + + "github.com/break/junhong_cmp_fiber/internal/handler/admin" + "github.com/break/junhong_cmp_fiber/internal/model/dto" + "github.com/break/junhong_cmp_fiber/pkg/constants" + "github.com/break/junhong_cmp_fiber/pkg/errors" + "github.com/break/junhong_cmp_fiber/pkg/middleware" + "github.com/break/junhong_cmp_fiber/pkg/openapi" +) + +// registerStopResumeRuleRoutes 注册停复机规则演练路由。 +// 门禁挂在独立前缀 /stop-resume-rule 的分组上，只作用于本组接口：代理、企业与个人客户账号一律 403。 +func registerStopResumeRuleRoutes(router fiber.Router, handler *admin.StopResumeRuleHandler, doc *openapi.Generator, basePath string) { + group := router.Group("/stop-resume-rule", func(c *fiber.Ctx) error { + userType := middleware.GetUserTypeFromContext(c.UserContext()) + if userType != constants.UserTypeSuperAdmin && userType != constants.UserTypePlatform { + return errors.New(errors.CodeForbidden, constants.PlatformManagementForbiddenMessage) + } + return c.Next() + }) + path := basePath + "/stop-resume-rule" + + Register(group, doc, path, "GET", "/dry-run", handler.DryRun, RouteSpec{ + Summary: "停复机规则演练", + Description: "对全部开机与停机卡按停复机新规则只判断不执行（不调用网关、不写任何状态），不受停复机总开关影响。" + + "返回按运营商账户 × 当前停机原因 × 判断结果 × 原因的汇总，以及按卡ID升序的分页明细；" + + "可按运营商账户与停机原因筛选，可用 assume_control_on / assume_control_off 假设通道实名管控配置预演。" + + "全量扫描较慢，生产使用请按运营商账户筛选或使用导出。仅超级管理员与平台账号可用。", + Tags: []string{"停复机规则"}, + Input: new(dto.StopResumeRuleDryRunRequest), + Output: new(dto.StopResumeRuleDryRunResponse), + Auth: true, + }) + + Register(group, doc, path, "POST", "/dry-run/export", handler.Export, RouteSpec{ + Summary: "导出停复机规则演练明细", + Description: "创建异步导出任务（场景 stop_resume_rule_dry_run），一行对应一张卡的判断结果；筛选与假设配置同演练接口。仅超级管理员与平台账号可用。", + Tags: []string{"停复机规则"}, + Input: new(dto.StopResumeRuleDryRunExportRequest), + Output: new(dto.CreateExportTaskResponse), + Auth: true, + }) +} diff --git a/internal/service/device/service.go b/internal/service/device/service.go index 978874db..65944061 100644 --- a/internal/service/device/service.go +++ b/internal/service/device/service.go @@ -30,6 +30,7 @@ import ( type Service struct { stopResumeRecorder *gatewaylimit.StopResumeRecorder + terminalPolling *gatewaylimit.TerminalPollingDisabler db *gorm.DB redis *redis.Client deviceStore *postgres.DeviceStore @@ -1436,6 +1437,7 @@ func (s *Service) StopDevice(ctx context.Context, deviceID uint) (*dto.DeviceSus s.recordCardCommand(ctx, card, constants.StopResumeActionStop, constants.StopReasonManual, gwErr, time.Since(callStartedAt)) integrationID = observer.integration if gwErr != nil { + s.disablePollingOnTerminal(ctx, card, constants.StopResumeActionStop, gwErr) log.Error("设备停机-调网关停机失败", zap.Uint("device_id", deviceID), zap.String("iccid", card.ICCID), @@ -1582,6 +1584,7 @@ func (s *Service) StartDevice(ctx context.Context, deviceID uint) error { s.recordCardCommand(ctx, card, constants.StopResumeActionStart, "", gwErr, time.Since(callStartedAt)) integrationID = observer.integration if gwErr != nil { + s.disablePollingOnTerminal(ctx, card, constants.StopResumeActionStart, gwErr) log.Error("设备复机-调网关复机失败", zap.Uint("device_id", deviceID), zap.String("iccid", card.ICCID), @@ -1735,6 +1738,18 @@ func (s *Service) SetStopResumeRecorder(recorder *gatewaylimit.StopResumeRecorde s.stopResumeRecorder = recorder } +// SetTerminalPollingDisabler 注入终态关闭轮询处理器：设备下某张卡停复机命中"自动关闭轮询"规则时只关闭该卡轮询。 +func (s *Service) SetTerminalPollingDisabler(disabler *gatewaylimit.TerminalPollingDisabler) { + s.terminalPolling = disabler +} + +// disablePollingOnTerminal 设备下单卡停复机失败命中终态规则时关闭该卡轮询并清除轮询卡缓存。 +func (s *Service) disablePollingOnTerminal(ctx context.Context, card *model.IotCard, action string, callErr error) { + if s.terminalPolling.HandleCommandError(ctx, card, action, callErr) { + s.invalidatePollingCardCache(card.ID) + } +} + // recordCardCommand 写设备下单卡的停复机记录；停复机只发一次请求，尝试序号恒为 1。 func (s *Service) recordCardCommand(ctx context.Context, card *model.IotCard, action, reason string, callErr error, duration time.Duration) { if s.stopResumeRecorder == nil || card == nil { diff --git a/internal/service/iot_card/stop_resume_service.go b/internal/service/iot_card/stop_resume_service.go index 4e8c5700..ce29c9e5 100644 --- a/internal/service/iot_card/stop_resume_service.go +++ b/internal/service/iot_card/stop_resume_service.go @@ -58,6 +58,8 @@ type StopResumeService struct { channelThresholdGuard ChannelThresholdLockGuard // stopResumeRecorder 写每次上游停复机调用的停复机记录；未注入时不写。 stopResumeRecorder *gatewaylimit.StopResumeRecorder + // terminalPolling 停复机命中"自动关闭轮询"终态规则时关闭该卡轮询；未注入时按普通失败处理。 + terminalPolling *gatewaylimit.TerminalPollingDisabler } // SetStopResumeRecorder 注入停复机记录写入器。 @@ -65,6 +67,23 @@ func (s *StopResumeService) SetStopResumeRecorder(recorder *gatewaylimit.StopRes s.stopResumeRecorder = recorder } +// SetTerminalPollingDisabler 注入终态关闭轮询处理器。 +func (s *StopResumeService) SetTerminalPollingDisabler(disabler *gatewaylimit.TerminalPollingDisabler) { + s.terminalPolling = disabler +} + +// disablePollingOnTerminal 停复机上游失败命中"自动关闭轮询"规则时关闭该卡轮询并通知调度器；返回 true 表示不再重试。 +func (s *StopResumeService) disablePollingOnTerminal(ctx context.Context, card *model.IotCard, action string, callErr error) bool { + if !s.terminalPolling.HandleCommandError(ctx, card, action, callErr) { + return false + } + card.EnablePolling = false + if s.pollingCallback != nil { + s.pollingCallback.OnCardDisabled(ctx, card.ID) + } + return true +} + // SetObservationSeriesEventWriter 注入停复机成功观测序列 Outbox Writer。 func (s *StopResumeService) SetObservationSeriesEventWriter(db *gorm.DB, writer cardObservationApp.SeriesEventWriter) { s.db = db @@ -747,6 +766,9 @@ func (s *StopResumeService) stopCardWithRetry(ctx context.Context, card *model.I } lastErr = callErr + if s.disablePollingOnTerminal(ctx, card, constants.StopResumeActionStop, callErr) { + break + } s.logger.Warn("调用停机接口失败，准备重试", zap.Int("attempt", i+1), zap.String("iccid", card.ICCID), @@ -823,6 +845,9 @@ func (s *StopResumeService) resumeCardWithRetry(ctx context.Context, card *model } lastErr = callErr + if s.disablePollingOnTerminal(ctx, card, constants.StopResumeActionStart, callErr) { + break + } s.logger.Warn("调用复机接口失败，准备重试", zap.Int("attempt", i+1), zap.String("iccid", card.ICCID), diff --git a/internal/service/polling/gateway_rate_limit_service.go b/internal/service/polling/gateway_rate_limit_service.go index aa027493..40240da1 100644 --- a/internal/service/polling/gateway_rate_limit_service.go +++ b/internal/service/polling/gateway_rate_limit_service.go @@ -491,7 +491,8 @@ func (s *GatewayRateLimitService) TestRule(ctx context.Context, req *dto.TestGat matched := result.Keyword != "" return &dto.TestGatewayErrorRuleResponse{ Category: result.Category, CategoryName: constants.GatewayErrorClassName(result.Category), - RuleID: result.RuleID, Keyword: result.Keyword, MaybeAccepted: result.MaybeAccepted, Matched: matched, + RuleID: result.RuleID, Keyword: result.Keyword, MaybeAccepted: result.MaybeAccepted, + AutoDisablePolling: result.AutoDisablePolling, Matched: matched, }, nil } @@ -510,6 +511,9 @@ func validateRule(req *dto.SaveGatewayErrorRuleRequest) error { if strings.TrimSpace(req.Keyword) == "" { return errors.New(errors.CodeInvalidParam, "匹配关键字不能为空") } + if req.AutoDisablePolling != nil && *req.AutoDisablePolling && req.Category != constants.GatewayClassTerminal { + return errors.New(errors.CodeInvalidParam, "只有终态不可操作类别的规则可以勾选自动关闭轮询") + } if req.MatchType == constants.GatewayMatchRegex { if _, err := regexp.Compile(req.Keyword); err != nil { return errors.New(errors.CodeInvalidParam, "正则表达式不合法") @@ -528,6 +532,13 @@ func applyRuleRequest(row *model.GatewayErrorClassification, req *dto.SaveGatewa row.Priority = 100 } row.MaybeAccepted = req.MaybeAccepted + if req.AutoDisablePolling != nil { + row.AutoDisablePolling = *req.AutoDisablePolling + } + // 类别改为非终态时自动关闭轮询标记随之失效，避免违反"仅终态可置位"。 + if row.Category != constants.GatewayClassTerminal { + row.AutoDisablePolling = false + } row.Remark = strings.TrimSpace(req.Remark) if req.Status != nil { row.Status = *req.Status @@ -537,7 +548,8 @@ func applyRuleRequest(row *model.GatewayErrorClassification, req *dto.SaveGatewa func ruleState(row *model.GatewayErrorClassification) map[string]any { ops, _ := gatewaylimit.DecodeOperations(row.Operations) return map[string]any{"keyword": row.Keyword, "match_type": row.MatchType, "category": row.Category, - "operations": ops, "priority": row.Priority, "maybe_accepted": row.MaybeAccepted, "status": row.Status} + "operations": ops, "priority": row.Priority, "maybe_accepted": row.MaybeAccepted, + "auto_disable_polling": row.AutoDisablePolling, "status": row.Status} } func ruleAudit(action, summary string, row *model.GatewayErrorClassification, operatorID uint, before map[string]any) auditinfra.PollingInput { @@ -556,7 +568,7 @@ func toRuleResponse(row *model.GatewayErrorClassification) dto.GatewayErrorRuleR return dto.GatewayErrorRuleResponse{ ID: row.ID, Keyword: row.Keyword, MatchType: row.MatchType, Category: row.Category, CategoryName: constants.GatewayErrorClassName(row.Category), Operations: nonNil(ops), Priority: row.Priority, - MaybeAccepted: row.MaybeAccepted, Status: row.Status, Remark: row.Remark, CreatedAt: row.CreatedAt, UpdatedAt: row.UpdatedAt, + MaybeAccepted: row.MaybeAccepted, AutoDisablePolling: row.AutoDisablePolling, Status: row.Status, Remark: row.Remark, CreatedAt: row.CreatedAt, UpdatedAt: row.UpdatedAt, } } diff --git a/internal/task/polling_cardstatus_handler.go b/internal/task/polling_cardstatus_handler.go index 5613bc36..9a3bafe4 100644 --- a/internal/task/polling_cardstatus_handler.go +++ b/internal/task/polling_cardstatus_handler.go @@ -29,7 +29,7 @@ func shouldStopPollingForRisk(card *model.IotCard, extend string) bool { if card == nil { return false } - return carddomain.ShouldStopPollingForRisk(card.IsStandalone, strings.TrimSpace(extend)) + return carddomain.ShouldStopPollingForRisk(strings.TrimSpace(extend)) } // NewPollingCardStatusHandler 创建卡状态轮询任务处理器。 @@ -133,7 +133,7 @@ func (h *PollingCardStatusHandler) Handle(ctx context.Context, task *asynq.Task) priority.Complete(ctx) h.base.updateStats(ctx, constants.TaskTypePollingCardStatus, true, time.Since(startedAt)) if decision.StopPolling { - h.base.logger.Info("独立卡命中风险状态，已关闭轮询", zap.Uint("card_id", cardID), zap.String("gateway_extend", decision.GatewayExtend)) + h.base.logger.Info("卡命中风险停机或已销户，已关闭该卡轮询", zap.Uint("card_id", cardID), zap.String("gateway_extend", decision.GatewayExtend)) return nil } diff --git a/migrations/000250_add_error_classification_auto_disable_polling.down.sql b/migrations/000250_add_error_classification_auto_disable_polling.down.sql new file mode 100644 index 00000000..d7c7ba8f --- /dev/null +++ b/migrations/000250_add_error_classification_auto_disable_polling.down.sql @@ -0,0 +1,3 @@ +-- 回滚：删除自动关闭轮询标记与约束，与 up 倒序成对。 +ALTER TABLE tb_gateway_error_classification DROP CONSTRAINT IF EXISTS ck_gateway_error_classification_auto_disable; +ALTER TABLE tb_gateway_error_classification DROP COLUMN IF EXISTS auto_disable_polling; diff --git a/migrations/000250_add_error_classification_auto_disable_polling.up.sql b/migrations/000250_add_error_classification_auto_disable_polling.up.sql new file mode 100644 index 00000000..ea07e4ba --- /dev/null +++ b/migrations/000250_add_error_classification_auto_disable_polling.up.sql @@ -0,0 +1,18 @@ +-- 终态规则可标记自动关闭轮询（finalize-unified-stop-resume-rule）。 +-- 背景：停复机调用直接返回"已销户""仅支持风险复机"等终态时，只按失败处理并重试，卡继续参与轮询与自动停复机； +-- 风险停机与已销户的卡应关闭自身轮询。 +-- +-- 设计选择： +-- 1. auto_disable_polling 由规则配置驱动，不在代码里写死关键字；只允许 category = 'terminal' 的规则置位（约束兜底，接口同样校验）。 +-- 2. 初始数据只为"已销户""仅支持风险复机"两条终态规则置位；"号码封存"等其他终态不置位，按失败告警由人工判断。 +-- 3. 按 keyword + category 定位初始规则，后台已修改或删除的规则不受影响。 + +ALTER TABLE tb_gateway_error_classification ADD COLUMN auto_disable_polling BOOLEAN NOT NULL DEFAULT FALSE; +ALTER TABLE tb_gateway_error_classification ADD CONSTRAINT ck_gateway_error_classification_auto_disable + CHECK (auto_disable_polling = FALSE OR category = 'terminal'); + +COMMENT ON COLUMN tb_gateway_error_classification.auto_disable_polling IS '命中后自动关闭该卡轮询（不重试、回写网关扩展状态），仅终态不可操作类别可置位'; + +UPDATE tb_gateway_error_classification +SET auto_disable_polling = TRUE, updated_at = NOW() +WHERE category = 'terminal' AND keyword IN ('已销户', '仅支持风险复机') AND deleted_at IS NULL; diff --git a/openspec/changes/finalize-unified-stop-resume-rule/tasks.md b/openspec/changes/finalize-unified-stop-resume-rule/tasks.md index 2d47584d..5c8d87f9 100644 --- a/openspec/changes/finalize-unified-stop-resume-rule/tasks.md +++ b/openspec/changes/finalize-unified-stop-resume-rule/tasks.md @@ -1,28 +1,28 @@ ## 0. 前置确认 -- [ ] 0.1 确认 `rebuild-card-traffic-metering-and-stop-resume` 已归档并上线生产（总开关可仍为关闭）；第 1、3 个 Change 已上线。 +- [x] 0.1 确认 `rebuild-card-traffic-metering-and-stop-resume` 已归档并上线生产（总开关可仍为关闭）；第 1、3 个 Change 已上线。（按用户 2026-10-24 决定以测试环境为准：`rebuild-card-traffic-metering-and-stop-resume` 代码与迁移已在测试环境，第 1、3 个 Change 已部署测试环境；生产前置由维护者发布时确认） ## 第一次发布：演练与口径补齐（总开关仍关闭，行为不变） ## 1. 迁移 -- [ ] 1.1 新增成对迁移：`tb_gateway_error_classification` 增加 `auto_disable_polling boolean not null default false`，为"已销户""风险停机"类终态规则置位；down 删列。 +- [x] 1.1 新增成对迁移：`tb_gateway_error_classification` 增加 `auto_disable_polling boolean not null default false`，为"已销户""风险停机"类终态规则置位；down 删列。（`000250_add_error_classification_auto_disable_polling`，含"仅终态可置位"约束） ## 2. 规则演练 -- [ ] 2.1 新增演练 Query：按批读取卡与资产数据，复用新规则判断函数（不受总开关影响，始终按新规则判断），支持假设的实名管控配置；只读。 -- [ ] 2.2 `GET /stop-resume-rule/dry-run`（运营商账户、停机原因筛选，汇总 + 分页明细），仅超级管理员与平台用户；导出场景 `stop_resume_rule_dry_run`。 -- [ ] 2.3 同步路由与文档生成入口，重新生成接口文档。 +- [x] 2.1 新增演练 Query：按批读取卡与资产数据，复用新规则判断函数（不受总开关影响，始终按新规则判断），支持假设的实名管控配置；只读。（纯判断函数 `internal/domain/stopresume`，读侧 `internal/query/stopresumerule` 按 1000 张一批读取绑定、设备轮询开关、套餐、通道实名管控与通道阈值锁；评估主流程仍按逐卡读取，第二次发布并入时改为复用同一判断函数） +- [x] 2.2 `GET /stop-resume-rule/dry-run`（运营商账户、停机原因筛选，汇总 + 分页明细），仅超级管理员与平台用户；导出场景 `stop_resume_rule_dry_run`。（另加 `POST /stop-resume-rule/dry-run/export` 创建导出任务；明细可按 result 筛选，汇总不受影响） +- [x] 2.3 同步路由与文档生成入口，重新生成接口文档。 ## 3. 风险停机与已销户关闭轮询 -- [ ] 3.1 `ShouldStopPollingForRisk` 去掉独立卡限制，只关闭该卡自身 `enable_polling`，不级联设备。 -- [ ] 3.2 停复机调用失败时按错误分类查 `auto_disable_polling`：置位则不重试、关闭该卡轮询、回写 `gateway_extend`、写审计。 -- [ ] 3.3 错误分类接口增加 `auto_disable_polling` 字段，非终态类别勾选时拒绝。 +- [x] 3.1 `ShouldStopPollingForRisk` 去掉独立卡限制，只关闭该卡自身 `enable_polling`，不级联设备。 +- [x] 3.2 停复机调用失败时按错误分类查 `auto_disable_polling`：置位则不重试、关闭该卡轮询、回写 `gateway_extend`、写审计。（`gatewaylimit.TerminalPollingDisabler`，卡级停复机与设备停复机共用；扩展状态按返回信息推断"已销户/风险停机"，推断不出保留原值；新增审计动作 `iot_card.terminal_disable_polling`） +- [x] 3.3 错误分类接口增加 `auto_disable_polling` 字段，非终态类别勾选时拒绝。（更新请求不传该字段时保留原值；类别改为非终态时自动清除） ## 4. 第一次发布的编译与验证 -- [ ] 4.1 `gofmt -w`；`go build ./cmd/api ./cmd/worker` 通过；`openspec validate finalize-unified-stop-resume-rule` 通过。 +- [x] 4.1 `gofmt -w`；`go build ./cmd/api ./cmd/worker` 通过；`openspec validate finalize-unified-stop-resume-rule` 通过。 - [ ] 4.2 测试环境：演练不改任何状态；开启总开关后抽取 `not_realname`、`carrier_stopped`、`manual`、停因为空各若干张，实际评估结果与演练一致。 - [ ] 4.3 测试环境：绑定设备的卡返回已销户时只关该卡轮询；复机返回风险停机时不重试并关闭轮询；号码封存不关闭轮询。 diff --git a/pkg/constants/audit.go b/pkg/constants/audit.go index 1d5683de..209c4996 100644 --- a/pkg/constants/audit.go +++ b/pkg/constants/audit.go @@ -139,6 +139,8 @@ const ( AuditActionIotCardDeactivated = "iot_card.deactivate" // AuditActionIotCardPollingStatusUpdated 表示更新 IoT 卡轮询开关。 AuditActionIotCardPollingStatusUpdated = "iot_card.update_polling_status" + // AuditActionIotCardTerminalPollingDisabled 表示停复机调用命中"自动关闭轮询"终态规则后系统关闭该卡轮询。 + AuditActionIotCardTerminalPollingDisabled = "iot_card.terminal_disable_polling" // AuditActionIotCardPollingStatusBatchUpdated 表示批量更新 IoT 卡轮询开关。 AuditActionIotCardPollingStatusBatchUpdated = "iot_card.batch_update_polling_status" // AuditActionIotCardBatchDeleted 表示批量删除 IoT 卡根事件。 diff --git a/pkg/constants/constants.go b/pkg/constants/constants.go index 7c80f4c3..e897f696 100644 --- a/pkg/constants/constants.go +++ b/pkg/constants/constants.go @@ -409,6 +409,8 @@ const ( ExportTaskSceneOperationsActivation = "operations_activation" // ExportTaskSceneOperationsRenewal 表示套餐续费情况报表导出场景，导出行与页面分组行同口径并含合计行。 ExportTaskSceneOperationsRenewal = "operations_renewal" + // ExportTaskSceneStopResumeRuleDryRun 表示停复机规则演练明细导出场景，一行对应一张卡的判断结果。 + ExportTaskSceneStopResumeRuleDryRun = "stop_resume_rule_dry_run" ) // ExportTaskInvalidTimeFilterMessage 是导出任务冻结的时间边界非法时写入 error_message 的安全失败摘要。