feat: 停复机规则演练与终态自动关闭轮询
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 9m26s

统一停复机规则转正的第一次发布:总开关仍关闭,现有停复机行为不变。

- 新增停复机规则演练:GET /stop-resume-rule/dry-run 按新规则对开机与停机卡只判断不执行,返回运营商账户 × 停机原因 × 判断结果汇总与分页明细,可假设通道实名管控配置;POST /stop-resume-rule/dry-run/export 导出明细(场景 stop_resume_rule_dry_run);仅超级管理员与平台账号
- 风险停机、已销户关闭轮询不再限于独立卡,只关该卡自身轮询,不级联设备
- 错误分类规则增加 auto_disable_polling,仅终态类别可勾选;停复机(含设备停复机)命中该规则时不重试、关闭该卡轮询、回写网关扩展状态并写审计 iot_card.terminal_disable_polling
- 迁移 000250 加列与约束,并为"已销户""仅支持风险复机"置位
This commit is contained in:
邹进
2026-10-25 17:52:40 +09:00
parent fad7dbfdff
commit af8249158d
33 changed files with 1227 additions and 56 deletions

View File

@@ -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

View File

@@ -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),

View File

@@ -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)

View File

@@ -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

View File

@@ -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)

View File

@@ -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
}

View File

@@ -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
}
}

View File

@@ -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

View File

@@ -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)
}

View File

@@ -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)
}

View File

@@ -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,

View File

@@ -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}
}

View File

@@ -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 任务因限流被推迟;同一任务首次被限流起累计推迟超过阈值时写一次错误告警,任务继续重试。

View File

@@ -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
}
}

View File

@@ -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:"任务状态名称(中文)"`

View File

@@ -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 停复机记录列表请求

View File

@@ -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:"每页数量"`
}

View File

@@ -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 返回错误分类规则表名。

View File

@@ -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)
}

View File

@@ -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
}

View File

@@ -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)
}

View File

@@ -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,
})
}

View File

@@ -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 {

View File

@@ -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),

View File

@@ -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,
}
}

View File

@@ -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
}