From c741f1751d483e5d8a02156ffb6f4101a438d795 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=82=B9=E8=BF=9B?= Date: Sun, 25 Oct 2026 18:56:32 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8C=89=E8=BF=90=E8=90=A5=E5=95=86?= =?UTF-8?q?=E8=B4=A6=E6=88=B7=E8=B0=83=E5=BA=A6=E7=9A=84=E5=90=8E=E5=8F=B0?= =?UTF-8?q?=E6=8E=A5=E5=8F=A3=E4=B8=8E=E9=87=8D=E7=AE=97=E5=88=B0=E6=9C=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - /polling-carrier-params 列表（含默认标记与消费能力）、详情、保存（执行上限不得大于订阅进程 max_in_flight 之和）、删除；/polling-check-scenes 列表与修改（检查类型、递增的确认延迟、冷却）；/polling-failure-alerts 列表、按账户与错误类别汇总、详情、标记已处理；GET /iot-cards/:id/polling-state 单卡轮询状态；GET /polling-stats/carrier-queues 按账户队列与执行监控 - 读取仅超级管理员与平台账号，修改仅超级管理员（告警处理平台账号可用），均写统一审计；新增审计动作与资源 - carrier 模式下重算到期时间写入卡到期信息与账户队列，只提前不推后 --- cmd/api/docs.go | 1 + cmd/gendocs/main.go | 1 + cmd/worker/main.go | 6 +- internal/bootstrap/handlers.go | 18 +- internal/bootstrap/types.go | 1 + internal/handler/admin/carrier_scheduler.go | 195 ++++++ internal/infrastructure/audit/registry.go | 20 + internal/model/dto/carrier_polling_dto.go | 185 ++++++ internal/polling/carrier_bridge.go | 27 + internal/polling/recalculate.go | 36 + internal/routes/admin.go | 3 + internal/routes/carrier_scheduler.go | 68 ++ .../polling/carrier_scheduler_service.go | 622 ++++++++++++++++++ pkg/constants/audit.go | 14 + 14 files changed, 1190 insertions(+), 7 deletions(-) create mode 100644 internal/handler/admin/carrier_scheduler.go create mode 100644 internal/model/dto/carrier_polling_dto.go create mode 100644 internal/routes/carrier_scheduler.go create mode 100644 internal/service/polling/carrier_scheduler_service.go diff --git a/cmd/api/docs.go b/cmd/api/docs.go index 41c19b3a..b78e5011 100644 --- a/cmd/api/docs.go +++ b/cmd/api/docs.go @@ -33,6 +33,7 @@ func generateOpenAPIDocs(outputPath string, logger *zap.Logger) { handlers.PhoneAssetAssociation = admin.NewPhoneAssetAssociationHandler(nil, nil) // 网关限流 Handler 必须同时进入文档工厂，避免新增管理接口遗漏文档注册。 handlers.GatewayRateLimit = admin.NewGatewayRateLimitHandler(nil, nil, nil) + handlers.CarrierScheduler = admin.NewCarrierSchedulerHandler(nil, nil) // 套餐真流量预警 Handler 必须同时进入文档工厂，避免新增管理接口遗漏文档注册。 handlers.PackageTrafficAlert = admin.NewPackageTrafficAlertHandler(nil, nil, nil, nil) // 资产钱包自动续费配置 Handler 必须同时进入文档工厂，避免新增管理接口遗漏文档注册。 diff --git a/cmd/gendocs/main.go b/cmd/gendocs/main.go index 4578428a..b2a75b73 100644 --- a/cmd/gendocs/main.go +++ b/cmd/gendocs/main.go @@ -42,6 +42,7 @@ func generateAdminDocs(outputPath string) error { handlers.PhoneAssetAssociation = admin.NewPhoneAssetAssociationHandler(nil, nil) // 网关限流 Handler 必须同时进入文档工厂，避免新增管理接口遗漏文档注册。 handlers.GatewayRateLimit = admin.NewGatewayRateLimitHandler(nil, nil, nil) + handlers.CarrierScheduler = admin.NewCarrierSchedulerHandler(nil, nil) // 套餐真流量预警 Handler 必须同时进入文档工厂，避免新增管理接口遗漏文档注册。 handlers.PackageTrafficAlert = admin.NewPackageTrafficAlertHandler(nil, nil, nil, nil) // 资产钱包自动续费配置 Handler 必须同时进入文档工厂，避免新增管理接口遗漏文档注册。 diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 94604a5d..f84a5b09 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -196,9 +196,9 @@ func runWorker(cfg *config.Config) { taskHandler.GetMux().HandleFunc(constants.TaskTypeOutboxDeliver, outboxHandler.Handle) partitionMaintainer := gatewaylimit.NewPartitionMaintainer(postgres.NewCardStopResumeRecordStore(runtime.db), runtime.gatewayLimit.Settings, appLogger) taskHandler.GetMux().HandleFunc(constants.TaskTypeStopResumeRecordPartition, partitionMaintainer.Handle) - recalcHandler := polling.NewRecalculateTaskHandler( - polling.NewDueRecalculator(runtime.db, runtime.redisClient, runtime.pollingConfigMgr, runtime.pollingQueueMgr, appLogger), - runtime.redisClient, appLogger) + dueRecalculator := polling.NewDueRecalculator(runtime.db, runtime.redisClient, runtime.pollingConfigMgr, runtime.pollingQueueMgr, appLogger) + dueRecalculator.SetCarrierScheduling(runtime.carrierScheduling) + recalcHandler := polling.NewRecalculateTaskHandler(dueRecalculator, runtime.redisClient, appLogger) taskHandler.GetMux().HandleFunc(constants.TaskTypePollingRecalculateDue, recalcHandler.Handle) startOutboxRelay(ctx, runtime, cfg.Worker.InstanceName, appLogger) rescuePendingImportTasks(ctx, runtime, appLogger) diff --git a/internal/bootstrap/handlers.go b/internal/bootstrap/handlers.go index cc2811d9..2ac6c7bd 100644 --- a/internal/bootstrap/handlers.go +++ b/internal/bootstrap/handlers.go @@ -337,10 +337,12 @@ func initHandlers(svc *services, deps *Dependencies) *Handlers { PollingConfig: admin.NewPollingConfigHandler(svc.PollingConfig), GatewayRateLimit: admin.NewGatewayRateLimitHandler(pollingSvcPkg.NewGatewayRateLimitService(deps.DB, auditInfra.NewWriter(auditInfra.NewRegistry(), nil), deps.GatewayLimit), deps.Redis, validate), PollingConcurrency: admin.NewPollingConcurrencyHandler(svc.PollingConcurrency), - PollingMonitoring: admin.NewPollingMonitoringHandler(svc.PollingMonitoring), - PollingAlert: admin.NewPollingAlertHandler(svc.PollingAlert), - PollingCleanup: admin.NewPollingCleanupHandler(svc.PollingCleanup), - PollingManualTrigger: admin.NewPollingManualTriggerHandler(svc.PollingManualTrigger), + CarrierScheduler: admin.NewCarrierSchedulerHandler(pollingSvcPkg.NewCarrierSchedulerService(deps.DB, deps.Redis, + auditInfra.NewWriter(auditInfra.NewRegistry(), nil), carrierSchedulerConfig()), validate), + PollingMonitoring: admin.NewPollingMonitoringHandler(svc.PollingMonitoring), + PollingAlert: admin.NewPollingAlertHandler(svc.PollingAlert), + PollingCleanup: admin.NewPollingCleanupHandler(svc.PollingCleanup), + PollingManualTrigger: admin.NewPollingManualTriggerHandler(svc.PollingManualTrigger), PollingPriority: admin.NewPriorityPollingHandler( svc.PriorityPolling, priorityPollingQuery.NewQuery(deps.DB, pollingPriorityItemStore), validate, ), @@ -408,3 +410,11 @@ func initHandlers(svc *services, deps *Dependencies) *Handlers { AgentOpenAPI: openapiHandler.NewHandler(svc.AgentOpenAPI, validate), } } + +// carrierSchedulerConfig 返回按运营商账户调度的 yaml 参数（供后台校验执行上限与展示默认值）。 +func carrierSchedulerConfig() config.CarrierSchedulerConfig { + if cfg := config.Get(); cfg != nil { + return cfg.Polling.Carrier + } + return config.CarrierSchedulerConfig{} +} diff --git a/internal/bootstrap/types.go b/internal/bootstrap/types.go index dc495112..917bfecf 100644 --- a/internal/bootstrap/types.go +++ b/internal/bootstrap/types.go @@ -65,6 +65,7 @@ type Handlers struct { WeComApprovalCallback *callback.WeComApprovalHandler PollingConfig *admin.PollingConfigHandler GatewayRateLimit *admin.GatewayRateLimitHandler + CarrierScheduler *admin.CarrierSchedulerHandler PollingConcurrency *admin.PollingConcurrencyHandler PollingMonitoring *admin.PollingMonitoringHandler PollingAlert *admin.PollingAlertHandler diff --git a/internal/handler/admin/carrier_scheduler.go b/internal/handler/admin/carrier_scheduler.go new file mode 100644 index 00000000..c7de430c --- /dev/null +++ b/internal/handler/admin/carrier_scheduler.go @@ -0,0 +1,195 @@ +package admin + +import ( + "strings" + + "github.com/go-playground/validator/v10" + "github.com/gofiber/fiber/v2" + + "github.com/break/junhong_cmp_fiber/internal/model/dto" + pollingSvc "github.com/break/junhong_cmp_fiber/internal/service/polling" + "github.com/break/junhong_cmp_fiber/pkg/errors" + "github.com/break/junhong_cmp_fiber/pkg/response" +) + +// CarrierSchedulerHandler 按运营商账户调度的后台接口：调度参数、请求检查场景、失败告警、单卡轮询状态、按账户监控。 +type CarrierSchedulerHandler struct { + service *pollingSvc.CarrierSchedulerService + validator *validator.Validate +} + +// NewCarrierSchedulerHandler 创建按运营商账户调度 Handler。 +func NewCarrierSchedulerHandler(service *pollingSvc.CarrierSchedulerService, validator *validator.Validate) *CarrierSchedulerHandler { + return &CarrierSchedulerHandler{service: service, validator: validator} +} + +func (h *CarrierSchedulerHandler) validate(prefix string, req any) error { + if err := h.validator.Struct(req); err != nil { + return errors.New(errors.CodeInvalidParam, validationMessage(prefix, req, err)) + } + return nil +} + +// ListParams 查询运营商账户调度参数列表。GET /api/admin/polling-carrier-params +func (h *CarrierSchedulerHandler) ListParams(c *fiber.Ctx) error { + result, err := h.service.ListParams(c.UserContext()) + if err != nil { + return err + } + return response.Success(c, result) +} + +// GetParam 查询运营商账户调度参数。GET /api/admin/polling-carrier-params/:carrier_id +func (h *CarrierSchedulerHandler) GetParam(c *fiber.Ctx) error { + carrierID, err := rateLimitPathID(c, "carrier_id", "运营商账户 ID") + if err != nil { + return err + } + result, err := h.service.GetParam(c.UserContext(), carrierID) + if err != nil { + return err + } + return response.Success(c, result) +} + +// SaveParam 保存运营商账户调度参数。PUT /api/admin/polling-carrier-params/:carrier_id +func (h *CarrierSchedulerHandler) SaveParam(c *fiber.Ctx) error { + var req dto.SaveCarrierPollingParamRequest + if err := c.BodyParser(&req); err != nil { + return errors.New(errors.CodeInvalidParam, "请求参数解析失败") + } + carrierID, err := rateLimitPathID(c, "carrier_id", "运营商账户 ID") + if err != nil { + return err + } + req.CarrierID = carrierID + if err := h.validate("运营商账户调度参数不合法", &req); err != nil { + return err + } + result, err := h.service.SaveParam(c.UserContext(), &req) + if err != nil { + return err + } + return response.Success(c, result) +} + +// DeleteParam 删除运营商账户调度参数（恢复全局默认）。DELETE /api/admin/polling-carrier-params/:carrier_id +func (h *CarrierSchedulerHandler) DeleteParam(c *fiber.Ctx) error { + carrierID, err := rateLimitPathID(c, "carrier_id", "运营商账户 ID") + if err != nil { + return err + } + if err := h.service.DeleteParam(c.UserContext(), carrierID); err != nil { + return err + } + return response.Success(c, nil) +} + +// ListScenes 查询请求检查场景。GET /api/admin/polling-check-scenes +func (h *CarrierSchedulerHandler) ListScenes(c *fiber.Ctx) error { + result, err := h.service.ListScenes(c.UserContext()) + if err != nil { + return err + } + return response.Success(c, result) +} + +// UpdateScene 修改请求检查场景。PUT /api/admin/polling-check-scenes/:scene +func (h *CarrierSchedulerHandler) UpdateScene(c *fiber.Ctx) error { + var req dto.UpdatePollingCheckSceneRequest + if err := c.BodyParser(&req); err != nil { + return errors.New(errors.CodeInvalidParam, "请求参数解析失败") + } + req.Scene = strings.TrimSpace(c.Params("scene")) + if req.Scene == "" { + return errors.New(errors.CodeInvalidParam, "无效的场景编码") + } + if err := h.validate("请求检查场景参数不合法", &req); err != nil { + return err + } + result, err := h.service.UpdateScene(c.UserContext(), &req) + if err != nil { + return err + } + return response.Success(c, result) +} + +// ListAlerts 查询卡轮询失败告警列表。GET /api/admin/polling-failure-alerts +func (h *CarrierSchedulerHandler) ListAlerts(c *fiber.Ctx) error { + var req dto.PollingFailureAlertListRequest + if err := c.QueryParser(&req); err != nil { + return errors.New(errors.CodeInvalidParam, "请求参数解析失败") + } + if err := h.validate("卡轮询失败告警查询参数不合法", &req); err != nil { + return err + } + result, err := h.service.ListAlerts(c.UserContext(), &req) + if err != nil { + return err + } + return response.Success(c, result) +} + +// AlertSummary 未处理告警按运营商账户 × 错误类别汇总。GET /api/admin/polling-failure-alerts/summary +func (h *CarrierSchedulerHandler) AlertSummary(c *fiber.Ctx) error { + result, err := h.service.AlertSummary(c.UserContext()) + if err != nil { + return err + } + return response.Success(c, result) +} + +// GetAlert 查询卡轮询失败告警详情。GET /api/admin/polling-failure-alerts/:id +func (h *CarrierSchedulerHandler) GetAlert(c *fiber.Ctx) error { + id, err := rateLimitPathID(c, "id", "告警 ID") + if err != nil { + return err + } + result, err := h.service.GetAlert(c.UserContext(), id) + if err != nil { + return err + } + return response.Success(c, result) +} + +// HandleAlert 标记卡轮询失败告警已处理。PUT /api/admin/polling-failure-alerts/:id/status +func (h *CarrierSchedulerHandler) HandleAlert(c *fiber.Ctx) error { + var req dto.HandlePollingFailureAlertRequest + if err := c.BodyParser(&req); err != nil { + return errors.New(errors.CodeInvalidParam, "请求参数解析失败") + } + id, err := rateLimitPathID(c, "id", "告警 ID") + if err != nil { + return err + } + req.ID = id + if err := h.validate("卡轮询失败告警处理参数不合法", &req); err != nil { + return err + } + if err := h.service.HandleAlert(c.UserContext(), &req); err != nil { + return err + } + return response.Success(c, nil) +} + +// CardState 查询单卡轮询状态。GET /api/admin/iot-cards/:id/polling-state +func (h *CarrierSchedulerHandler) CardState(c *fiber.Ctx) error { + id, err := rateLimitPathID(c, "id", "卡 ID") + if err != nil { + return err + } + result, err := h.service.CardState(c.UserContext(), id) + if err != nil { + return err + } + return response.Success(c, result) +} + +// CarrierStats 按运营商账户的轮询队列与执行监控。GET /api/admin/polling-stats/carrier-queues +func (h *CarrierSchedulerHandler) CarrierStats(c *fiber.Ctx) error { + result, err := h.service.CarrierStats(c.UserContext()) + 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 caa5a2f4..171d144f 100644 --- a/internal/infrastructure/audit/registry.go +++ b/internal/infrastructure/audit/registry.go @@ -277,6 +277,10 @@ func NewRegistry() *Registry { gatewayErrorRuleCreated := pollingAction(constants.AuditActionGatewayErrorRuleCreated, "创建上游错误分类规则", constants.AuditResourceGatewayErrorRule, constants.AuditRiskNormal) gatewayErrorRuleUpdated := pollingAction(constants.AuditActionGatewayErrorRuleUpdated, "更新上游错误分类规则", constants.AuditResourceGatewayErrorRule, constants.AuditRiskNormal) gatewayErrorRuleStatusUpdated := pollingAction(constants.AuditActionGatewayErrorRuleStatusUpdated, "更新上游错误分类规则状态", constants.AuditResourceGatewayErrorRule, constants.AuditRiskNormal) + carrierPollingParamSaved := pollingAction(constants.AuditActionCarrierPollingParamSaved, "保存运营商账户调度参数", constants.AuditResourceCarrierPollingParam, constants.AuditRiskHigh) + carrierPollingParamDeleted := pollingAction(constants.AuditActionCarrierPollingParamDeleted, "删除运营商账户调度参数", constants.AuditResourceCarrierPollingParam, constants.AuditRiskHigh) + pollingCheckSceneUpdated := pollingAction(constants.AuditActionPollingCheckSceneUpdated, "修改请求检查场景", constants.AuditResourcePollingCheckScene, constants.AuditRiskNormal) + pollingFailureAlertHandled := pollingAction(constants.AuditActionPollingFailureAlertHandled, "处理卡轮询失败告警", constants.AuditResourcePollingFailureAlert, constants.AuditRiskNormal) gatewayErrorRuleDeleted := pollingAction(constants.AuditActionGatewayErrorRuleDeleted, "删除上游错误分类规则", constants.AuditResourceGatewayErrorRule, constants.AuditRiskNormal) pollingConcurrencyUpdated := pollingAction(constants.AuditActionPollingConcurrencyUpdated, "更新轮询并发配置", constants.AuditResourcePollingConcurrencyConfig, constants.AuditRiskNormal) pollingConcurrencyReset := pollingAction(constants.AuditActionPollingConcurrencyReset, "重置轮询并发计数", constants.AuditResourcePollingConcurrencyConfig, constants.AuditRiskNormal) @@ -678,6 +682,10 @@ func NewRegistry() *Registry { constants.AuditActionGatewayErrorRuleUpdated: gatewayErrorRuleUpdated, constants.AuditActionGatewayErrorRuleStatusUpdated: gatewayErrorRuleStatusUpdated, constants.AuditActionGatewayErrorRuleDeleted: gatewayErrorRuleDeleted, + constants.AuditActionCarrierPollingParamSaved: carrierPollingParamSaved, + constants.AuditActionCarrierPollingParamDeleted: carrierPollingParamDeleted, + constants.AuditActionPollingCheckSceneUpdated: pollingCheckSceneUpdated, + constants.AuditActionPollingFailureAlertHandled: pollingFailureAlertHandled, constants.AuditActionPollingConcurrencyUpdated: pollingConcurrencyUpdated, constants.AuditActionPollingConcurrencyReset: pollingConcurrencyReset, constants.AuditActionPollingAlertRuleCreated: pollingAlertRuleCreated, @@ -966,6 +974,18 @@ func NewRegistry() *Registry { Type: constants.AuditResourceGatewayErrorRule, Name: "上游错误分类规则", IdentityFields: []string{"id", "keyword", "match_type", "category", "priority", "status"}, }, + constants.AuditResourceCarrierPollingParam: { + Type: constants.AuditResourceCarrierPollingParam, Name: "运营商账户调度参数", + IdentityFields: []string{"carrier_id", "inflight_limit", "card_max_in_flight", "priority_max_ratio"}, + }, + constants.AuditResourcePollingCheckScene: { + Type: constants.AuditResourcePollingCheckScene, Name: "请求检查场景", + IdentityFields: []string{"scene", "scene_name"}, + }, + constants.AuditResourcePollingFailureAlert: { + Type: constants.AuditResourcePollingFailureAlert, Name: "卡轮询失败告警", + IdentityFields: []string{"id", "alert_level", "card_id", "carrier_id", "check_type", "status"}, + }, constants.AuditResourcePollingAlertRule: { Type: constants.AuditResourcePollingAlertRule, Name: "轮询告警规则", IdentityFields: []string{"id", "rule_name", "task_type", "metric_type", "operator", "threshold", "alert_level", "status"}, diff --git a/internal/model/dto/carrier_polling_dto.go b/internal/model/dto/carrier_polling_dto.go new file mode 100644 index 00000000..da3a9873 --- /dev/null +++ b/internal/model/dto/carrier_polling_dto.go @@ -0,0 +1,185 @@ +package dto + +import "time" + +// CarrierPollingParamItem 运营商账户调度参数（含全局默认合并后的生效值）。 +type CarrierPollingParamItem struct { + CarrierID uint `json:"carrier_id" description:"运营商账户ID"` + CarrierName string `json:"carrier_name" description:"运营商账户名称"` + IsDefault bool `json:"is_default" description:"是否未单独配置、全部使用全局默认"` + InflightLimit int `json:"inflight_limit" description:"生效的执行上限（跨 Worker 实例同时执行卡任务数）"` + CardMaxInFlight int `json:"card_max_in_flight" description:"生效的每进程卡任务 max_in_flight"` + PriorityMaxRatio *float64 `json:"priority_max_ratio" description:"优先队列每轮最大占比（0~1），空表示不限"` + StopResumeMaxInFlight *int `json:"stop_resume_max_in_flight" description:"每进程停复机 max_in_flight（停复机指令队列使用），空表示默认"` + SRInflightLimit *int `json:"sr_inflight_limit" description:"停复机执行上限（停复机指令队列使用），空表示默认"` + StopResumeMinInterval *int `json:"stop_resume_min_interval" description:"单卡两次停复机调用最小间隔（秒），空表示默认"` + StopResumeConfirmDelay *int `json:"stop_resume_confirm_delay" description:"停复机结果未知后确认延迟（秒），空表示默认"` + ConfiguredInflight *int `json:"configured_inflight_limit" description:"单独配置的执行上限，空表示使用默认"` + ConfiguredMaxInFlight *int `json:"configured_card_max_in_flight" description:"单独配置的每进程 max_in_flight，空表示使用默认"` + ConsumeCapacity int `json:"consume_capacity" description:"当前订阅进程 max_in_flight 之和（执行上限不应超过该值）"` +} + +// CarrierPollingParamListResponse 运营商账户调度参数列表。 +type CarrierPollingParamListResponse struct { + Items []CarrierPollingParamItem `json:"items" description:"各运营商账户调度参数"` + SubscriberProcesses int `json:"subscriber_processes" description:"订阅卡任务的 Worker 进程数"` + DefaultInflight int `json:"default_inflight_limit" description:"全局默认执行上限"` + DefaultMaxInFlight int `json:"default_card_max_in_flight" description:"全局默认每进程 max_in_flight"` +} + +// CarrierPollingParamPath 运营商账户路径参数。 +type CarrierPollingParamPath struct { + CarrierID uint `json:"-" path:"carrier_id" required:"true" description:"运营商账户ID"` +} + +// SaveCarrierPollingParamRequest 保存运营商账户调度参数；字段为空表示该项使用全局默认。 +type SaveCarrierPollingParamRequest struct { + CarrierID uint `json:"-" path:"carrier_id" required:"true" description:"运营商账户ID"` + InflightLimit *int `json:"inflight_limit" validate:"omitempty,min=1,max=10000" minimum:"1" maximum:"10000" description:"执行上限，空表示默认"` + CardMaxInFlight *int `json:"card_max_in_flight" validate:"omitempty,min=1,max=2000" minimum:"1" maximum:"2000" description:"每进程卡任务 max_in_flight，空表示默认"` + StopResumeMaxInFlight *int `json:"stop_resume_max_in_flight" validate:"omitempty,min=1,max=1000" minimum:"1" maximum:"1000" description:"每进程停复机 max_in_flight，空表示默认"` + SRInflightLimit *int `json:"sr_inflight_limit" validate:"omitempty,min=1,max=10000" minimum:"1" maximum:"10000" description:"停复机执行上限，空表示默认"` + PriorityMaxRatio *float64 `json:"priority_max_ratio" validate:"omitempty,gt=0,lte=1" description:"优先队列每轮最大占比（0~1]，空表示不限"` + StopResumeMinInterval *int `json:"stop_resume_min_interval" validate:"omitempty,min=1,max=86400" minimum:"1" maximum:"86400" description:"单卡停复机最小间隔（秒），空表示默认"` + StopResumeConfirmDelay *int `json:"stop_resume_confirm_delay" validate:"omitempty,min=1,max=86400" minimum:"1" maximum:"86400" description:"停复机结果未知确认延迟（秒），空表示默认"` +} + +// PollingCheckSceneItem 请求检查场景。 +type PollingCheckSceneItem struct { + Scene string `json:"scene" description:"场景编码 (order_paid:订单支付, package_activated:套餐激活, stop_resume_succeeded:停复机成功, device_operation:设备操作, auto_renewal_purchase:自动续费购包, client_asset_detail:C 端资产详情, admin_asset_realtime:后台资产实时状态, realname_link:实名链接)"` + SceneName string `json:"scene_name" description:"场景名称"` + CheckTypes []string `json:"check_types" description:"检查类型 (card_status:卡状态, carddata:流量, realname:实名)"` + ConfirmDelays []int `json:"confirm_delays" description:"延迟确认序列（秒），空表示只立即检查一次"` + Cooldown *int `json:"cooldown" description:"冷却时长（秒），最近一次成功检查未超过该时长则不请求；空表示不冷却"` + UpdatedAt time.Time `json:"updated_at" description:"更新时间"` +} + +// UpdatePollingCheckSceneRequest 修改请求检查场景（场景由代码预置，只可改检查类型、确认延迟与冷却）。 +type UpdatePollingCheckSceneRequest struct { + Scene string `json:"-" path:"scene" required:"true" description:"场景编码"` + CheckTypes []string `json:"check_types" validate:"required,min=1,max=3,dive,oneof=card_status carddata realname" required:"true" description:"检查类型 (card_status:卡状态, carddata:流量, realname:实名)"` + ConfirmDelays []int `json:"confirm_delays" validate:"omitempty,max=5,dive,min=1,max=86400" description:"延迟确认序列（秒，递增），空表示只立即检查一次"` + Cooldown *int `json:"cooldown" validate:"omitempty,min=1,max=86400" minimum:"1" maximum:"86400" description:"冷却时长（秒），空表示不冷却"` +} + +// PollingFailureAlertListRequest 卡轮询失败告警列表请求。 +type PollingFailureAlertListRequest 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:"每页数量"` + CarrierID *uint `json:"carrier_id" query:"carrier_id" validate:"omitempty,min=1" description:"运营商账户ID"` + CheckType string `json:"check_type" query:"check_type" validate:"omitempty,oneof=card_status carddata realname package all" description:"检查类型 (card_status:卡状态, carddata:流量, realname:实名, package:套餐, all:账户级)"` + ErrorCategory string `json:"error_category" query:"error_category" validate:"omitempty,max=50" description:"错误类别"` + Status *int `json:"status" query:"status" validate:"omitempty,oneof=1 2 3 4" description:"状态 (1:未处理, 2:已恢复, 3:已关闭轮询, 4:已处理)"` + AlertLevel string `json:"alert_level" query:"alert_level" validate:"omitempty,oneof=card carrier" description:"告警级别 (card:卡级, carrier:账户级)"` + ICCID string `json:"iccid" query:"iccid" validate:"omitempty,max=30" description:"ICCID（精确）"` + StartTime string `json:"start_time" query:"start_time" description:"最近失败时间起始（带时区的 RFC3339 秒级时间）"` + EndTime string `json:"end_time" query:"end_time" description:"最近失败时间结束（带时区的 RFC3339 秒级时间）"` +} + +// PollingFailureAlertItem 卡轮询失败告警。 +type PollingFailureAlertItem struct { + ID uint `json:"id" description:"告警ID"` + AlertLevel string `json:"alert_level" description:"告警级别 (card:卡级, carrier:账户级)"` + CardID uint `json:"card_id" description:"卡ID，账户级为 0"` + ICCID string `json:"iccid" description:"ICCID"` + CarrierID uint `json:"carrier_id" description:"运营商账户ID"` + CarrierName string `json:"carrier_name" description:"运营商账户名称"` + CheckType string `json:"check_type" description:"检查类型"` + CheckTypeName string `json:"check_type_name" description:"检查类型名称"` + ErrorCategory string `json:"error_category" description:"最近错误类别"` + FailCount int `json:"fail_count" description:"连续失败次数；账户级为窗口内告警数"` + FirstFailedAt *time.Time `json:"first_failed_at" description:"首次失败时间"` + LastFailedAt *time.Time `json:"last_failed_at" description:"最近失败时间"` + LastError string `json:"last_error" description:"最近错误"` + Status int `json:"status" description:"状态 (1:未处理, 2:已恢复, 3:已关闭轮询, 4:已处理)"` + StatusName string `json:"status_name" description:"状态名称"` + HandledBy uint `json:"handled_by" description:"处理人"` + HandledAt *time.Time `json:"handled_at" description:"处理时间"` + HandleRemark string `json:"handle_remark" description:"处理备注"` + RecoveredAt *time.Time `json:"recovered_at" description:"自动恢复时间"` + CreatedAt time.Time `json:"created_at" description:"创建时间"` +} + +// PollingFailureAlertPageResult 卡轮询失败告警分页结果。 +type PollingFailureAlertPageResult struct { + List []PollingFailureAlertItem `json:"items" description:"告警列表"` + Total int64 `json:"total" description:"总数"` + Page int `json:"page" description:"当前页"` + PageSize int `json:"size" description:"每页数量"` +} + +// PollingFailureAlertSummaryItem 未处理告警按运营商账户 × 错误类别汇总。 +type PollingFailureAlertSummaryItem struct { + CarrierID uint `json:"carrier_id" description:"运营商账户ID"` + CarrierName string `json:"carrier_name" description:"运营商账户名称"` + ErrorCategory string `json:"error_category" description:"错误类别"` + OpenCount int64 `json:"open_count" description:"未处理卡级告警数"` +} + +// PollingFailureAlertIDPath 告警路径参数。 +type PollingFailureAlertIDPath struct { + ID uint `json:"-" path:"id" required:"true" description:"告警ID"` +} + +// HandlePollingFailureAlertRequest 标记告警已处理。 +type HandlePollingFailureAlertRequest struct { + ID uint `json:"-" path:"id" required:"true" description:"告警ID"` + Status int `json:"status" validate:"required,oneof=4" required:"true" description:"目标状态 (4:已处理)；关闭轮询请使用资产轮询开关接口"` + Remark string `json:"remark" validate:"omitempty,max=500" maxLength:"500" description:"处理备注"` +} + +// CardPollingStatePath 卡ID路径参数。 +type CardPollingStatePath struct { + ID uint `json:"-" path:"id" required:"true" description:"卡ID"` +} + +// CardPollingTypeState 卡某检查类型的轮询状态。 +type CardPollingTypeState struct { + CheckType string `json:"check_type" description:"检查类型"` + CheckTypeName string `json:"check_type_name" description:"检查类型名称"` + NextDueAt *time.Time `json:"next_due_at" description:"下次到期时间"` + LastSuccessAt *time.Time `json:"last_success_at" description:"最近一次成功时间"` + FailCount int `json:"fail_count" description:"连续失败次数"` + FirstFailedAt *time.Time `json:"first_failed_at" description:"首次失败时间"` + LastError string `json:"last_error" description:"最近错误（类别|信息）"` + PendingConfirm []int `json:"pending_confirm" description:"待执行的延迟确认（秒）"` +} + +// CardPollingStateResponse 单卡轮询状态（只读）。 +type CardPollingStateResponse struct { + CardID uint `json:"card_id" description:"卡ID"` + ICCID string `json:"iccid" description:"ICCID"` + CarrierID uint `json:"carrier_id" description:"运营商账户ID"` + SchedulerMode string `json:"scheduler_mode" description:"当前调度模式 (legacy:旧分片调度, carrier:按运营商账户调度)"` + EnablePolling bool `json:"enable_polling" description:"卡轮询开关"` + Queue string `json:"queue" description:"所在队列 (priority:优先, queue:普通, none:不在轮询中)"` + QueueScoreAt *time.Time `json:"queue_score_at" description:"队列中下次可领取时间"` + Running bool `json:"running" description:"是否执行中"` + ClaimExpireAt *time.Time `json:"claim_expire_at" description:"处理中标记到期时间"` + PriorityMark bool `json:"priority_mark" description:"执行中收到请求检查，收尾后进入优先队列"` + Types []CardPollingTypeState `json:"types" description:"各检查类型状态"` +} + +// CarrierPollingStatsItem 按运营商账户的轮询监控。 +type CarrierPollingStatsItem struct { + CarrierID uint `json:"carrier_id" description:"运营商账户ID"` + CarrierName string `json:"carrier_name" description:"运营商账户名称"` + PriorityLength int64 `json:"priority_length" description:"优先队列长度"` + QueueLength int64 `json:"queue_length" description:"普通队列长度"` + DueCount int64 `json:"due_count" description:"已到期卡数（优先 + 普通）"` + Running int64 `json:"running" description:"执行中卡任务数"` + InflightLimit int `json:"inflight_limit" description:"执行上限"` + QPSLastMinute float64 `json:"qps_last_minute" description:"上一分钟实际网关调用 QPS（查询类）"` + RateLimitWaits int64 `json:"rate_limit_waits" description:"上一分钟限流等待次数"` + RateLimited int64 `json:"rate_limited" description:"上一分钟限流拒绝或等待超时次数"` + ExpiredClaims1h int64 `json:"expired_claims_1h" description:"近 1 小时处理中标记到期次数（执行崩溃或超时）"` + OpenAlerts int64 `json:"open_alerts" description:"未处理卡级失败告警数"` + OpenAlertsByType map[string]int64 `json:"open_alerts_by_category" description:"未处理卡级失败告警数（按错误类别）"` +} + +// CarrierPollingStatsResponse 按运营商账户的轮询监控。 +type CarrierPollingStatsResponse struct { + SchedulerMode string `json:"scheduler_mode" description:"当前调度模式"` + InitState string `json:"init_state" description:"新结构初始化状态 (running / done，空表示未初始化)"` + Items []CarrierPollingStatsItem `json:"items" description:"各运营商账户监控"` +} diff --git a/internal/polling/carrier_bridge.go b/internal/polling/carrier_bridge.go index dc8edaa5..e4f8d12c 100644 --- a/internal/polling/carrier_bridge.go +++ b/internal/polling/carrier_bridge.go @@ -2,10 +2,12 @@ package polling import ( "context" + "strconv" "sync" "sync/atomic" "time" + "github.com/redis/go-redis/v9" "go.uber.org/zap" "github.com/break/junhong_cmp_fiber/internal/model" @@ -197,3 +199,28 @@ func (i *CarrierInitializer) setProgress(update func(*InitProgress)) { update(&i.progress) i.mu.Unlock() } + +// AdvanceDue 把卡某检查类型的到期时间提前到 due（只提前不推后），并同步把账户队列中的 score 提前；返回是否提前。 +// 卡不在轮询中（无到期信息）时不处理。 +func (c *CarrierScheduling) AdvanceDue(ctx context.Context, card *model.IotCard, checkType string, due time.Time) (bool, error) { + if c == nil || c.store == nil || card == nil { + return false, nil + } + current, err := c.store.ReadDue(ctx, card.ID) + if err != nil || current.CarrierID == 0 { + return false, err + } + item, ok := current.Types[checkType] + if !ok || !item.HasDue || !due.Before(item.Due) { + return false, nil + } + client := c.store.Client() + member := strconv.FormatUint(uint64(card.ID), 10) + pipe := client.TxPipeline() + pipe.HSet(ctx, constants.RedisCardDueKey(card.ID), carrierqueue.FieldDue(checkType), due.UnixMilli()) + score := redis.ZAddArgs{XX: true, LT: true, Members: []redis.Z{{Score: float64(due.UnixMilli()), Member: member}}} + pipe.ZAddArgs(ctx, constants.RedisCarrierQueueKey(current.CarrierID), score) + pipe.ZAddArgs(ctx, constants.RedisCarrierPriorityQueueKey(current.CarrierID), score) + _, err = pipe.Exec(ctx) + return err == nil, err +} diff --git a/internal/polling/recalculate.go b/internal/polling/recalculate.go index d57897a3..fd13dea5 100644 --- a/internal/polling/recalculate.go +++ b/internal/polling/recalculate.go @@ -32,6 +32,39 @@ type DueRecalculator struct { configMgr *PollingConfigManager queueMgr *PollingQueueManager logger *zap.Logger + // carrier 按运营商账户调度接缝：carrier 模式下重算写入卡到期信息与账户队列。 + carrier *CarrierScheduling +} + +// SetCarrierScheduling 注入按运营商账户调度接缝。 +func (r *DueRecalculator) SetCarrierScheduling(carrier *CarrierScheduling) { + r.carrier = carrier +} + +// recalculateCarrierBatch 按运营商账户调度时重算：各检查类型候选到期同旧结构口径，只提前不推后。 +func (r *DueRecalculator) recalculateCarrierBatch(ctx context.Context, cards []*model.IotCard) (int, error) { + now := time.Now() + advanced := 0 + for _, card := range cards { + for taskType, info := range r.configMgr.MergedTaskIntervals(card) { + checkType := constants.CheckTypeFromTaskType(taskType) + if checkType == "" || info.Interval <= 0 { + continue + } + due := recalculatedDue(lastCheckAtByTaskType(card, taskType), info.Interval, now) + if windowDue := r.configMgr.NextCheckAt(ctx, card, taskType, info, now); windowDue.Before(due) && windowDue.After(now) { + due = windowDue + } + ok, err := r.carrier.AdvanceDue(ctx, card, checkType, due) + if err != nil { + return advanced, err + } + if ok { + advanced++ + } + } + } + return advanced, nil } // NewDueRecalculator 创建重算器。 @@ -97,6 +130,9 @@ func (r *DueRecalculator) recalculateBatch(ctx context.Context, cards []*model.I if tier := r.configMgr.TierResolver(); tier != nil { tier.Prefill(ctx, cards) } + if r.carrier.Active(ctx) { + return r.recalculateCarrierBatch(ctx, cards) + } now := time.Now() pipe := r.redis.Pipeline() var cmds []*redis.IntCmd diff --git a/internal/routes/admin.go b/internal/routes/admin.go index 765b8721..af702ef1 100644 --- a/internal/routes/admin.go +++ b/internal/routes/admin.go @@ -136,6 +136,9 @@ func RegisterAdminRoutes(router fiber.Router, handlers *bootstrap.Handlers, midd if handlers.PollingConfig != nil { registerPollingConfigRoutes(authGroup, handlers.PollingConfig, doc, basePath) } + if handlers.CarrierScheduler != nil { + registerCarrierSchedulerRoutes(authGroup, handlers.CarrierScheduler, doc, basePath) + } if handlers.GatewayRateLimit != nil { registerGatewayRateLimitRoutes(authGroup, handlers.GatewayRateLimit, doc, basePath) } diff --git a/internal/routes/carrier_scheduler.go b/internal/routes/carrier_scheduler.go new file mode 100644 index 00000000..264647b8 --- /dev/null +++ b/internal/routes/carrier_scheduler.go @@ -0,0 +1,68 @@ +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/openapi" +) + +// registerCarrierSchedulerRoutes 注册按运营商账户调度的后台接口。读取仅超级管理员与平台账号，修改仅超级管理员（服务层校验）。 +func registerCarrierSchedulerRoutes(router fiber.Router, handler *admin.CarrierSchedulerHandler, doc *openapi.Generator, basePath string) { + tags := []string{"轮询调度"} + + params := router.Group("/polling-carrier-params") + paramsPath := basePath + "/polling-carrier-params" + Register(params, doc, paramsPath, "GET", "", handler.ListParams, RouteSpec{ + Summary: "查询运营商账户调度参数列表", Description: "未单独配置的账户返回全局默认并标记 is_default。", Tags: tags, + Input: nil, Output: new(dto.CarrierPollingParamListResponse), Auth: true, + }) + Register(params, doc, paramsPath, "GET", "/:carrier_id", handler.GetParam, RouteSpec{ + Summary: "查询运营商账户调度参数", Tags: tags, Input: new(dto.CarrierPollingParamPath), Output: new(dto.CarrierPollingParamItem), Auth: true, + }) + Register(params, doc, paramsPath, "PUT", "/:carrier_id", handler.SaveParam, RouteSpec{ + Summary: "保存运营商账户调度参数（仅超级管理员）", + Description: "字段为空表示该项使用全局默认；执行上限不得大于当前订阅进程 max_in_flight 之和；下一轮调度生效并写统一审计。", + Tags: tags, Input: new(dto.SaveCarrierPollingParamRequest), Output: new(dto.CarrierPollingParamItem), Auth: true, + }) + Register(params, doc, paramsPath, "DELETE", "/:carrier_id", handler.DeleteParam, RouteSpec{ + Summary: "删除运营商账户调度参数（恢复全局默认，仅超级管理员）", Tags: tags, Input: new(dto.CarrierPollingParamPath), Output: nil, Auth: true, + }) + + scenes := router.Group("/polling-check-scenes") + scenesPath := basePath + "/polling-check-scenes" + Register(scenes, doc, scenesPath, "GET", "", handler.ListScenes, RouteSpec{ + Summary: "查询请求检查场景", Description: "业务事件与读入口在按运营商账户调度时按场景请求检查；场景由代码预置。", Tags: tags, + Input: nil, Output: new([]dto.PollingCheckSceneItem), Auth: true, + }) + Register(scenes, doc, scenesPath, "PUT", "/:scene", handler.UpdateScene, RouteSpec{ + Summary: "修改请求检查场景（仅超级管理员）", Description: "只可修改检查类型、延迟确认序列（递增）与冷却时长。", Tags: tags, + Input: new(dto.UpdatePollingCheckSceneRequest), Output: new(dto.PollingCheckSceneItem), Auth: true, + }) + + alerts := router.Group("/polling-failure-alerts") + alertsPath := basePath + "/polling-failure-alerts" + Register(alerts, doc, alertsPath, "GET", "", handler.ListAlerts, RouteSpec{ + Summary: "查询卡轮询失败告警列表", Tags: tags, Input: new(dto.PollingFailureAlertListRequest), Output: new(dto.PollingFailureAlertPageResult), Auth: true, + }) + Register(alerts, doc, alertsPath, "GET", "/summary", handler.AlertSummary, RouteSpec{ + Summary: "未处理卡轮询失败告警按运营商账户与错误类别汇总", Tags: tags, Input: nil, Output: new([]dto.PollingFailureAlertSummaryItem), Auth: true, + }) + Register(alerts, doc, alertsPath, "GET", "/:id", handler.GetAlert, RouteSpec{ + Summary: "查询卡轮询失败告警详情", Tags: tags, Input: new(dto.PollingFailureAlertIDPath), Output: new(dto.PollingFailureAlertItem), Auth: true, + }) + Register(alerts, doc, alertsPath, "PUT", "/:id/status", handler.HandleAlert, RouteSpec{ + Summary: "标记卡轮询失败告警已处理", Description: "只有未处理告警可标记；关闭该卡轮询请使用资产轮询开关接口。", Tags: tags, + Input: new(dto.HandlePollingFailureAlertRequest), Output: nil, Auth: true, + }) + + Register(router, doc, basePath, "GET", "/iot-cards/:id/polling-state", handler.CardState, RouteSpec{ + Summary: "查询单卡轮询状态", Description: "各检查类型下次到期、最近成功、连续失败与最近错误、是否执行中；只读。", Tags: tags, + Input: new(dto.CardPollingStatePath), Output: new(dto.CardPollingStateResponse), Auth: true, + }) + Register(router, doc, basePath, "GET", "/polling-stats/carrier-queues", handler.CarrierStats, RouteSpec{ + Summary: "按运营商账户的轮询队列与执行监控", Description: "按运营商账户调度的各账户队列长度、已到期数、执行中、上一分钟 QPS、限流等待与拒绝、处理中标记到期、未处理告警。", Tags: tags, + Input: nil, Output: new(dto.CarrierPollingStatsResponse), Auth: true, + }) +} diff --git a/internal/service/polling/carrier_scheduler_service.go b/internal/service/polling/carrier_scheduler_service.go new file mode 100644 index 00000000..53c5aaf2 --- /dev/null +++ b/internal/service/polling/carrier_scheduler_service.go @@ -0,0 +1,622 @@ +package polling + +import ( + "context" + "encoding/json" + stderrors "errors" + "strconv" + "time" + + "github.com/redis/go-redis/v9" + "gorm.io/datatypes" + "gorm.io/gorm" + + auditinfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/audit" + "github.com/break/junhong_cmp_fiber/internal/model" + "github.com/break/junhong_cmp_fiber/internal/model/dto" + "github.com/break/junhong_cmp_fiber/internal/polling/carrierqueue" + "github.com/break/junhong_cmp_fiber/pkg/config" + "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/utils" +) + +// CarrierSchedulerService 按运营商账户调度的后台维护与监控：调度参数、请求检查场景、失败告警、单卡轮询状态与按账户监控。 +// 读取仅超级管理员与平台账号；修改仅超级管理员并写统一审计，下一轮调度生效。 +type CarrierSchedulerService struct { + db *gorm.DB + redis *redis.Client + auditWriter *auditinfra.Writer + store *carrierqueue.Store + settings *carrierqueue.SettingsProvider + cfg config.CarrierSchedulerConfig +} + +// NewCarrierSchedulerService 创建按运营商账户调度后台服务。 +func NewCarrierSchedulerService(db *gorm.DB, redisClient *redis.Client, auditWriter *auditinfra.Writer, cfg config.CarrierSchedulerConfig) *CarrierSchedulerService { + return &CarrierSchedulerService{db: db, redis: redisClient, auditWriter: auditWriter, store: carrierqueue.NewStore(redisClient), + settings: carrierqueue.NewDefaultSettingsProvider(db, redisClient, nil), cfg: cfg} +} + +func (s *CarrierSchedulerService) processes() int { + if s.cfg.SubscriberProcesses > 0 { + return s.cfg.SubscriberProcesses + } + return 1 +} + +// ListParams 列出全部运营商账户的调度参数（未配置的账户返回全局默认并标记默认）。 +func (s *CarrierSchedulerService) ListParams(ctx context.Context) (*dto.CarrierPollingParamListResponse, error) { + if err := requirePlatformReader(ctx); err != nil { + return nil, err + } + names, err := s.carrierNameMap(ctx) + if err != nil { + return nil, err + } + var rows []model.CarrierPollingParam + if err := s.db.WithContext(ctx).Find(&rows).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询运营商账户调度参数失败") + } + byCarrier := make(map[uint]model.CarrierPollingParam, len(rows)) + for _, row := range rows { + byCarrier[row.CarrierID] = row + } + defaults := carrierqueue.MergeParams(model.CarrierPollingParam{}, false, s.cfg) + resp := &dto.CarrierPollingParamListResponse{SubscriberProcesses: s.processes(), + DefaultInflight: defaults.InflightLimit, DefaultMaxInFlight: defaults.CardMaxInFlight, Items: []dto.CarrierPollingParamItem{}} + for _, id := range sortedCarrierIDs(names) { + row, configured := byCarrier[id] + resp.Items = append(resp.Items, s.paramItem(id, names[id], row, configured)) + } + return resp, nil +} + +func (s *CarrierSchedulerService) paramItem(carrierID uint, name string, row model.CarrierPollingParam, configured bool) dto.CarrierPollingParamItem { + merged := carrierqueue.MergeParams(row, configured, s.cfg) + item := dto.CarrierPollingParamItem{CarrierID: carrierID, CarrierName: name, IsDefault: !configured, + InflightLimit: merged.InflightLimit, CardMaxInFlight: merged.CardMaxInFlight, ConsumeCapacity: merged.CardMaxInFlight * s.processes()} + if configured { + item.PriorityMaxRatio, item.StopResumeMaxInFlight, item.SRInflightLimit = row.PriorityMaxRatio, row.StopResumeMaxInFlight, row.SRInflightLimit + item.StopResumeMinInterval, item.StopResumeConfirmDelay = row.StopResumeMinInterval, row.StopResumeConfirmDelay + item.ConfiguredInflight, item.ConfiguredMaxInFlight = row.InflightLimit, row.CardMaxInFlight + } + return item +} + +// GetParam 查询单个运营商账户调度参数。 +func (s *CarrierSchedulerService) GetParam(ctx context.Context, carrierID uint) (*dto.CarrierPollingParamItem, error) { + if err := requirePlatformReader(ctx); err != nil { + return nil, err + } + carrier, err := s.loadCarrier(ctx, carrierID) + if err != nil { + return nil, err + } + row, configured, err := s.loadParam(ctx, carrierID) + if err != nil { + return nil, err + } + item := s.paramItem(carrierID, carrier.CarrierName, row, configured) + return &item, nil +} + +// SaveParam 保存运营商账户调度参数；执行上限不得大于当前订阅进程 max_in_flight 之和。 +func (s *CarrierSchedulerService) SaveParam(ctx context.Context, req *dto.SaveCarrierPollingParamRequest) (*dto.CarrierPollingParamItem, error) { + operatorID, err := requireSuperAdminWriter(ctx) + if err != nil { + return nil, err + } + carrier, err := s.loadCarrier(ctx, req.CarrierID) + if err != nil { + return nil, err + } + row, configured, err := s.loadParam(ctx, req.CarrierID) + if err != nil { + return nil, err + } + before := paramState(row, configured) + row.CarrierID = req.CarrierID + row.InflightLimit, row.CardMaxInFlight, row.StopResumeMaxInFlight = req.InflightLimit, req.CardMaxInFlight, req.StopResumeMaxInFlight + row.SRInflightLimit, row.PriorityMaxRatio = req.SRInflightLimit, req.PriorityMaxRatio + row.StopResumeMinInterval, row.StopResumeConfirmDelay = req.StopResumeMinInterval, req.StopResumeConfirmDelay + merged := carrierqueue.MergeParams(row, true, s.cfg) + if capacity := merged.CardMaxInFlight * s.processes(); merged.InflightLimit > capacity { + return nil, errors.New(errors.CodeInvalidParam, "执行上限 "+strconv.Itoa(merged.InflightLimit)+" 大于当前订阅进程 max_in_flight 之和 "+ + strconv.Itoa(capacity)+"（"+strconv.Itoa(s.processes())+" 个进程 × "+strconv.Itoa(merged.CardMaxInFlight)+"），请调低执行上限或调高 max_in_flight") + } + if !configured { + row.Creator = operatorID + } + row.Updater = operatorID + err = runPollingTransaction(ctx, s.db, s.auditWriter, func(tx *gorm.DB) error { + if err := tx.WithContext(ctx).Save(&row).Error; err != nil { + return errors.Wrap(errors.CodeDatabaseError, err, "保存运营商账户调度参数失败") + } + return writePollingAudit(ctx, tx, s.auditWriter, auditinfra.PollingInput{ + ActionCode: constants.AuditActionCarrierPollingParamSaved, Summary: "保存运营商账户调度参数", + ResourceType: constants.AuditResourceCarrierPollingParam, ResourceID: req.CarrierID, + ResourceKey: "carrier_polling_param:" + uintString(req.CarrierID), DisplayName: carrier.CarrierName, OperatorID: operatorID, + IdentitySnapshot: map[string]any{"carrier_id": req.CarrierID, "inflight_limit": row.InflightLimit, + "card_max_in_flight": row.CardMaxInFlight, "priority_max_ratio": row.PriorityMaxRatio}, + BeforeData: before, AfterData: paramState(row, true), + }) + }) + if err != nil { + return nil, err + } + item := s.paramItem(req.CarrierID, carrier.CarrierName, row, true) + return &item, nil +} + +// DeleteParam 删除运营商账户调度参数，该账户恢复使用全局默认。 +func (s *CarrierSchedulerService) DeleteParam(ctx context.Context, carrierID uint) error { + operatorID, err := requireSuperAdminWriter(ctx) + if err != nil { + return err + } + row, configured, err := s.loadParam(ctx, carrierID) + if err != nil { + return err + } + if !configured { + return errors.New(errors.CodeNotFound, "该运营商账户未单独配置调度参数") + } + return runPollingTransaction(ctx, s.db, s.auditWriter, func(tx *gorm.DB) error { + if err := tx.WithContext(ctx).Delete(&model.CarrierPollingParam{}, row.ID).Error; err != nil { + return errors.Wrap(errors.CodeDatabaseError, err, "删除运营商账户调度参数失败") + } + return writePollingAudit(ctx, tx, s.auditWriter, auditinfra.PollingInput{ + ActionCode: constants.AuditActionCarrierPollingParamDeleted, Summary: "删除运营商账户调度参数（恢复全局默认）", + ResourceType: constants.AuditResourceCarrierPollingParam, ResourceID: carrierID, + ResourceKey: "carrier_polling_param:" + uintString(carrierID), DisplayName: uintString(carrierID), OperatorID: operatorID, + IdentitySnapshot: map[string]any{"carrier_id": carrierID}, BeforeData: paramState(row, true), + }) + }) +} + +func paramState(row model.CarrierPollingParam, configured bool) map[string]any { + if !configured { + return map[string]any{"configured": false} + } + return map[string]any{"configured": true, "inflight_limit": row.InflightLimit, "card_max_in_flight": row.CardMaxInFlight, + "stop_resume_max_in_flight": row.StopResumeMaxInFlight, "sr_inflight_limit": row.SRInflightLimit, + "priority_max_ratio": row.PriorityMaxRatio, "stop_resume_min_interval": row.StopResumeMinInterval, + "stop_resume_confirm_delay": row.StopResumeConfirmDelay} +} + +func (s *CarrierSchedulerService) loadParam(ctx context.Context, carrierID uint) (model.CarrierPollingParam, bool, error) { + var row model.CarrierPollingParam + err := s.db.WithContext(ctx).Where("carrier_id = ?", carrierID).First(&row).Error + if stderrors.Is(err, gorm.ErrRecordNotFound) { + return model.CarrierPollingParam{}, false, nil + } + if err != nil { + return row, false, errors.Wrap(errors.CodeDatabaseError, err, "查询运营商账户调度参数失败") + } + return row, true, nil +} + +func (s *CarrierSchedulerService) loadCarrier(ctx context.Context, carrierID uint) (*model.Carrier, error) { + var carrier model.Carrier + if err := s.db.WithContext(ctx).Select("id", "carrier_name").First(&carrier, carrierID).Error; err != nil { + if stderrors.Is(err, gorm.ErrRecordNotFound) { + return nil, errors.New(errors.CodeNotFound, "运营商账户不存在") + } + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询运营商账户失败") + } + return &carrier, nil +} + +func (s *CarrierSchedulerService) carrierNameMap(ctx context.Context) (map[uint]string, error) { + var carriers []model.Carrier + if err := s.db.WithContext(ctx).Select("id", "carrier_name").Find(&carriers).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询运营商账户失败") + } + names := make(map[uint]string, len(carriers)) + for _, carrier := range carriers { + names[carrier.ID] = carrier.CarrierName + } + return names, nil +} + +// ListScenes 列出请求检查场景。 +func (s *CarrierSchedulerService) ListScenes(ctx context.Context) ([]dto.PollingCheckSceneItem, error) { + if err := requirePlatformReader(ctx); err != nil { + return nil, err + } + var rows []model.PollingCheckScene + if err := s.db.WithContext(ctx).Order("id").Find(&rows).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询请求检查场景失败") + } + items := make([]dto.PollingCheckSceneItem, 0, len(rows)) + for _, row := range rows { + items = append(items, sceneItem(row)) + } + return items, nil +} + +func sceneItem(row model.PollingCheckScene) dto.PollingCheckSceneItem { + item := dto.PollingCheckSceneItem{Scene: row.Scene, SceneName: row.SceneName, Cooldown: row.Cooldown, UpdatedAt: row.UpdatedAt, + CheckTypes: []string{}, ConfirmDelays: []int{}} + _ = json.Unmarshal(row.CheckTypes, &item.CheckTypes) + _ = json.Unmarshal(row.ConfirmDelays, &item.ConfirmDelays) + return item +} + +// UpdateScene 修改请求检查场景的检查类型、确认延迟与冷却；场景由代码预置，不可新增或删除。 +func (s *CarrierSchedulerService) UpdateScene(ctx context.Context, req *dto.UpdatePollingCheckSceneRequest) (*dto.PollingCheckSceneItem, error) { + operatorID, err := requireSuperAdminWriter(ctx) + if err != nil { + return nil, err + } + for index := 1; index < len(req.ConfirmDelays); index++ { + if req.ConfirmDelays[index] <= req.ConfirmDelays[index-1] { + return nil, errors.New(errors.CodeInvalidParam, "延迟确认序列必须递增") + } + } + types := uniqueStrings(req.CheckTypes) + var row model.PollingCheckScene + if err := s.db.WithContext(ctx).Where("scene = ?", req.Scene).First(&row).Error; err != nil { + if stderrors.Is(err, gorm.ErrRecordNotFound) { + return nil, errors.New(errors.CodeNotFound, "请求检查场景不存在") + } + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询请求检查场景失败") + } + before := sceneItem(row) + typesJSON, _ := json.Marshal(types) + delays := req.ConfirmDelays + if delays == nil { + delays = []int{} + } + delaysJSON, _ := json.Marshal(delays) + row.CheckTypes, row.ConfirmDelays, row.Cooldown, row.Updater = datatypes.JSON(typesJSON), datatypes.JSON(delaysJSON), req.Cooldown, operatorID + err = runPollingTransaction(ctx, s.db, s.auditWriter, func(tx *gorm.DB) error { + if err := tx.WithContext(ctx).Model(&model.PollingCheckScene{}).Where("id = ?", row.ID).Updates(map[string]any{ + "check_types": row.CheckTypes, "confirm_delays": row.ConfirmDelays, "cooldown": row.Cooldown, "updater": operatorID, "updated_at": time.Now(), + }).Error; err != nil { + return errors.Wrap(errors.CodeDatabaseError, err, "修改请求检查场景失败") + } + return writePollingAudit(ctx, tx, s.auditWriter, auditinfra.PollingInput{ + ActionCode: constants.AuditActionPollingCheckSceneUpdated, Summary: "修改请求检查场景", + ResourceType: constants.AuditResourcePollingCheckScene, ResourceID: row.ID, + ResourceKey: "polling_check_scene:" + row.Scene, DisplayName: row.SceneName, OperatorID: operatorID, + IdentitySnapshot: map[string]any{"scene": row.Scene, "scene_name": row.SceneName}, + BeforeData: map[string]any{"check_types": before.CheckTypes, "confirm_delays": before.ConfirmDelays, "cooldown": before.Cooldown}, + AfterData: map[string]any{"check_types": types, "confirm_delays": delays, "cooldown": req.Cooldown}, + }) + }) + if err != nil { + return nil, err + } + item := sceneItem(row) + return &item, nil +} + +// ListAlerts 分页查询卡轮询失败告警。 +func (s *CarrierSchedulerService) ListAlerts(ctx context.Context, req *dto.PollingFailureAlertListRequest) (*dto.PollingFailureAlertPageResult, error) { + if err := requirePlatformReader(ctx); err != nil { + return nil, err + } + page, size := normalizePage(req.Page, req.PageSize) + query := s.db.WithContext(ctx).Model(&model.PollingFailureAlert{}) + if req.CarrierID != nil { + query = query.Where("carrier_id = ?", *req.CarrierID) + } + if req.CheckType != "" { + query = query.Where("check_type = ?", req.CheckType) + } + if req.ErrorCategory != "" { + query = query.Where("error_category = ?", req.ErrorCategory) + } + if req.Status != nil { + query = query.Where("status = ?", *req.Status) + } + if req.AlertLevel != "" { + query = query.Where("alert_level = ?", req.AlertLevel) + } + if req.ICCID != "" { + query = query.Where("iccid = ?", req.ICCID) + } + start, end, err := utils.ParseTimeRange(req.StartTime, req.EndTime) + if err != nil { + return nil, err + } + if start != nil { + query = query.Where("last_failed_at >= ?", *start) + } + if end != nil { + query = query.Where("last_failed_at <= ?", *end) + } + var total int64 + if err := query.Count(&total).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "统计卡轮询失败告警失败") + } + var rows []model.PollingFailureAlert + if err := query.Order("last_failed_at DESC NULLS LAST, id DESC").Offset((page - 1) * size).Limit(size).Find(&rows).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询卡轮询失败告警失败") + } + names, err := s.carrierNameMap(ctx) + if err != nil { + return nil, err + } + result := &dto.PollingFailureAlertPageResult{Total: total, Page: page, PageSize: size, List: make([]dto.PollingFailureAlertItem, 0, len(rows))} + for _, row := range rows { + result.List = append(result.List, alertItem(row, names)) + } + return result, nil +} + +func alertItem(row model.PollingFailureAlert, names map[uint]string) dto.PollingFailureAlertItem { + return dto.PollingFailureAlertItem{ + ID: row.ID, AlertLevel: row.AlertLevel, CardID: row.CardID, ICCID: row.ICCID, CarrierID: row.CarrierID, + CarrierName: names[row.CarrierID], CheckType: row.CheckType, CheckTypeName: constants.CheckTypeName(row.CheckType), + ErrorCategory: row.ErrorCategory, FailCount: row.FailCount, FirstFailedAt: row.FirstFailedAt, LastFailedAt: row.LastFailedAt, + LastError: row.LastError, Status: row.Status, StatusName: constants.PollingAlertStatusName(row.Status), + HandledBy: row.HandledBy, HandledAt: row.HandledAt, HandleRemark: row.HandleRemark, RecoveredAt: row.RecoveredAt, CreatedAt: row.CreatedAt, + } +} + +// AlertSummary 未处理卡级告警按运营商账户 × 错误类别汇总。 +func (s *CarrierSchedulerService) AlertSummary(ctx context.Context) ([]dto.PollingFailureAlertSummaryItem, error) { + if err := requirePlatformReader(ctx); err != nil { + return nil, err + } + var rows []struct { + CarrierID uint + ErrorCategory string + OpenCount int64 + } + if err := s.db.WithContext(ctx).Model(&model.PollingFailureAlert{}). + Select("carrier_id, error_category, COUNT(*) AS open_count"). + Where("status = ? AND alert_level = ?", constants.PollingAlertStatusOpen, constants.PollingAlertLevelCard). + Group("carrier_id, error_category").Order("open_count DESC").Scan(&rows).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "汇总卡轮询失败告警失败") + } + names, err := s.carrierNameMap(ctx) + if err != nil { + return nil, err + } + items := make([]dto.PollingFailureAlertSummaryItem, 0, len(rows)) + for _, row := range rows { + items = append(items, dto.PollingFailureAlertSummaryItem{CarrierID: row.CarrierID, CarrierName: names[row.CarrierID], + ErrorCategory: row.ErrorCategory, OpenCount: row.OpenCount}) + } + return items, nil +} + +// GetAlert 查询告警详情。 +func (s *CarrierSchedulerService) GetAlert(ctx context.Context, id uint) (*dto.PollingFailureAlertItem, error) { + if err := requirePlatformReader(ctx); err != nil { + return nil, err + } + row, err := s.loadAlert(ctx, id) + if err != nil { + return nil, err + } + names, err := s.carrierNameMap(ctx) + if err != nil { + return nil, err + } + item := alertItem(*row, names) + return &item, nil +} + +// HandleAlert 把未处理告警标记为已处理。 +func (s *CarrierSchedulerService) HandleAlert(ctx context.Context, req *dto.HandlePollingFailureAlertRequest) error { + operatorID, err := requireSuperAdminWriterOrPlatform(ctx) + if err != nil { + return err + } + row, err := s.loadAlert(ctx, req.ID) + if err != nil { + return err + } + if row.Status != constants.PollingAlertStatusOpen { + return errors.New(errors.CodeInvalidStatus, "只有未处理的告警可以标记已处理") + } + now := time.Now() + return runPollingTransaction(ctx, s.db, s.auditWriter, func(tx *gorm.DB) error { + result := tx.WithContext(ctx).Model(&model.PollingFailureAlert{}).Where("id = ? AND status = ?", row.ID, constants.PollingAlertStatusOpen). + Updates(map[string]any{"status": req.Status, "handled_by": operatorID, "handled_at": now, "handle_remark": req.Remark, "updated_at": now}) + if result.Error != nil { + return errors.Wrap(errors.CodeDatabaseError, result.Error, "处理卡轮询失败告警失败") + } + if result.RowsAffected == 0 { + return errors.New(errors.CodeConflict, "告警状态已变化，请刷新后重试") + } + return writePollingAudit(ctx, tx, s.auditWriter, auditinfra.PollingInput{ + ActionCode: constants.AuditActionPollingFailureAlertHandled, Summary: "标记卡轮询失败告警已处理", + ResourceType: constants.AuditResourcePollingFailureAlert, ResourceID: row.ID, + ResourceKey: "polling_failure_alert:" + uintString(row.ID), DisplayName: row.ICCID, OperatorID: operatorID, + IdentitySnapshot: map[string]any{"id": row.ID, "alert_level": row.AlertLevel, "card_id": row.CardID, + "carrier_id": row.CarrierID, "check_type": row.CheckType, "status": row.Status}, + BeforeData: map[string]any{"status": row.Status}, AfterData: map[string]any{"status": req.Status, "handle_remark": req.Remark}, + }) + }) +} + +func (s *CarrierSchedulerService) loadAlert(ctx context.Context, id uint) (*model.PollingFailureAlert, error) { + var row model.PollingFailureAlert + if err := s.db.WithContext(ctx).First(&row, id).Error; err != nil { + if stderrors.Is(err, gorm.ErrRecordNotFound) { + return nil, errors.New(errors.CodeNotFound, "卡轮询失败告警不存在") + } + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询卡轮询失败告警失败") + } + return &row, nil +} + +// CardState 查询单卡轮询状态（只读）。 +func (s *CarrierSchedulerService) CardState(ctx context.Context, cardID uint) (*dto.CardPollingStateResponse, error) { + if err := requirePlatformReader(ctx); err != nil { + return nil, err + } + var card model.IotCard + if err := s.db.WithContext(ctx).Select("id", "iccid", "carrier_id", "enable_polling").First(&card, cardID).Error; err != nil { + if stderrors.Is(err, gorm.ErrRecordNotFound) { + return nil, errors.New(errors.CodeNotFound, "IoT卡不存在") + } + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询IoT卡失败") + } + resp := &dto.CardPollingStateResponse{CardID: card.ID, ICCID: card.ICCID, CarrierID: card.CarrierID, + SchedulerMode: s.settings.Current(ctx).Mode, EnablePolling: card.EnablePolling, Queue: "none", Types: []dto.CardPollingTypeState{}} + due, err := s.store.ReadDue(ctx, card.ID) + if err != nil { + return nil, errors.Wrap(errors.CodeRedisError, err, "读取卡轮询到期信息失败") + } + carrierID := due.CarrierID + if carrierID == 0 { + carrierID = card.CarrierID + } + member := strconv.FormatUint(uint64(card.ID), 10) + if score, err := s.redis.ZScore(ctx, constants.RedisCarrierPriorityQueueKey(carrierID), member).Result(); err == nil { + resp.Queue, resp.QueueScoreAt = "priority", msTime(int64(score)) + } else if score, err := s.redis.ZScore(ctx, constants.RedisCarrierQueueKey(carrierID), member).Result(); err == nil { + resp.Queue, resp.QueueScoreAt = "queue", msTime(int64(score)) + } + if ttl, err := s.redis.PTTL(ctx, constants.RedisCardClaimKey(card.ID)).Result(); err == nil && ttl > 0 { + resp.Running = true + expire := time.Now().Add(ttl) + resp.ClaimExpireAt = &expire + } + resp.PriorityMark = due.Priority + for _, checkType := range constants.AllCheckTypes() { + item, ok := due.Types[checkType] + if !ok || (!item.HasDue && item.FailCount == 0 && item.LastOK.IsZero()) { + continue + } + state := dto.CardPollingTypeState{CheckType: checkType, CheckTypeName: constants.CheckTypeName(checkType), + FailCount: item.FailCount, LastError: item.LastError, PendingConfirm: []int{}} + if item.HasDue { + state.NextDueAt = timePtr(item.Due) + } + if !item.LastOK.IsZero() { + state.LastSuccessAt = timePtr(item.LastOK) + } + if !item.FirstFailed.IsZero() { + state.FirstFailedAt = timePtr(item.FirstFailed) + } + for _, delay := range item.Confirm { + state.PendingConfirm = append(state.PendingConfirm, int(delay/time.Second)) + } + resp.Types = append(resp.Types, state) + } + return resp, nil +} + +// CarrierStats 按运营商账户的轮询监控。 +func (s *CarrierSchedulerService) CarrierStats(ctx context.Context) (*dto.CarrierPollingStatsResponse, error) { + if err := requirePlatformReader(ctx); err != nil { + return nil, err + } + names, err := s.carrierNameMap(ctx) + if err != nil { + return nil, err + } + var params []model.CarrierPollingParam + if err := s.db.WithContext(ctx).Find(¶ms).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询运营商账户调度参数失败") + } + paramByCarrier := map[uint]model.CarrierPollingParam{} + for _, row := range params { + paramByCarrier[row.CarrierID] = row + } + var alertRows []struct { + CarrierID uint + ErrorCategory string + Count int64 + } + if err := s.db.WithContext(ctx).Model(&model.PollingFailureAlert{}).Select("carrier_id, error_category, COUNT(*) AS count"). + Where("status = ? AND alert_level = ?", constants.PollingAlertStatusOpen, constants.PollingAlertLevelCard). + Group("carrier_id, error_category").Scan(&alertRows).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "统计卡轮询失败告警失败") + } + state, _ := s.redis.Get(ctx, constants.RedisCarrierSchedulerInitStateKey).Result() + resp := &dto.CarrierPollingStatsResponse{SchedulerMode: s.settings.Current(ctx).Mode, InitState: state, Items: []dto.CarrierPollingStatsItem{}} + now := time.Now() + nowMS := strconv.FormatInt(now.UnixMilli(), 10) + lastMinute := now.Add(-time.Minute).In(time.FixedZone("CST", 8*3600)).Format("200601021504") + for _, carrierID := range sortedCarrierIDs(names) { + row, configured := paramByCarrier[carrierID] + item := dto.CarrierPollingStatsItem{CarrierID: carrierID, CarrierName: names[carrierID], + InflightLimit: carrierqueue.MergeParams(row, configured, s.cfg).InflightLimit, OpenAlertsByType: map[string]int64{}} + pipe := s.redis.Pipeline() + priorityLen := pipe.ZCard(ctx, constants.RedisCarrierPriorityQueueKey(carrierID)) + queueLen := pipe.ZCard(ctx, constants.RedisCarrierQueueKey(carrierID)) + priorityDue := pipe.ZCount(ctx, constants.RedisCarrierPriorityQueueKey(carrierID), "-inf", nowMS) + queueDue := pipe.ZCount(ctx, constants.RedisCarrierQueueKey(carrierID), "-inf", nowMS) + running := pipe.ZCount(ctx, constants.RedisCarrierRunningKey(carrierID), nowMS, "+inf") + expired := pipe.HGetAll(ctx, constants.RedisCarrierExpiredClaimsKey(carrierID)) + var statCmds []*redis.MapStringStringCmd + for _, op := range []string{constants.GatewayOpQueryStatus, constants.GatewayOpQueryFlow, constants.GatewayOpQueryRealname} { + statCmds = append(statCmds, pipe.HGetAll(ctx, constants.RedisGatewayCallStatKey(carrierID, op, lastMinute))) + } + _, _ = pipe.Exec(ctx) + item.PriorityLength, item.QueueLength = priorityLen.Val(), queueLen.Val() + item.DueCount, item.Running = priorityDue.Val()+queueDue.Val(), running.Val() + cutoff := now.Add(-time.Hour).Unix() / 60 + for minute, value := range expired.Val() { + if m, err := strconv.ParseInt(minute, 10, 64); err == nil && m >= cutoff { + count, _ := strconv.ParseInt(value, 10, 64) + item.ExpiredClaims1h += count + } + } + var attempts int64 + for _, cmd := range statCmds { + values := cmd.Val() + attempts += atoi64(values["attempts"]) + item.RateLimitWaits += atoi64(values["rate_wait"]) + item.RateLimited += atoi64(values[constants.GatewayErrorRateLimited]) + } + item.QPSLastMinute = float64(attempts) / 60 + for _, alert := range alertRows { + if alert.CarrierID == carrierID { + item.OpenAlerts += alert.Count + item.OpenAlertsByType[alert.ErrorCategory] += alert.Count + } + } + resp.Items = append(resp.Items, item) + } + return resp, nil +} + +// requireSuperAdminWriterOrPlatform 校验告警处理权限：超级管理员与平台账号均可，返回操作者 ID。 +func requireSuperAdminWriterOrPlatform(ctx context.Context) (uint, error) { + if err := requirePlatformReader(ctx); err != nil { + return 0, err + } + operatorID := middleware.GetUserIDFromContext(ctx) + if operatorID == 0 { + return 0, errors.New(errors.CodeUnauthorized, "未授权访问") + } + return operatorID, nil +} + +func sortedCarrierIDs(names map[uint]string) []uint { + ids := make([]uint, 0, len(names)) + for id := range names { + ids = append(ids, id) + } + for i := 1; i < len(ids); i++ { + for j := i; j > 0 && ids[j] < ids[j-1]; j-- { + ids[j], ids[j-1] = ids[j-1], ids[j] + } + } + return ids +} + +func msTime(ms int64) *time.Time { + t := time.UnixMilli(ms) + return &t +} + +func timePtr(t time.Time) *time.Time { + return &t +} + +func atoi64(s string) int64 { + value, _ := strconv.ParseInt(s, 10, 64) + return value +} diff --git a/pkg/constants/audit.go b/pkg/constants/audit.go index 209c4996..4a859c28 100644 --- a/pkg/constants/audit.go +++ b/pkg/constants/audit.go @@ -602,6 +602,14 @@ const ( AuditActionGatewayErrorRuleStatusUpdated = "gateway_error_rule.update_status" // AuditActionGatewayErrorRuleDeleted 表示删除上游错误分类规则。 AuditActionGatewayErrorRuleDeleted = "gateway_error_rule.delete" + // AuditActionCarrierPollingParamSaved 表示保存运营商账户调度参数。 + AuditActionCarrierPollingParamSaved = "carrier_polling_param.save" + // AuditActionCarrierPollingParamDeleted 表示删除运营商账户调度参数（恢复全局默认）。 + AuditActionCarrierPollingParamDeleted = "carrier_polling_param.delete" + // AuditActionPollingCheckSceneUpdated 表示修改请求检查场景配置。 + AuditActionPollingCheckSceneUpdated = "polling_check_scene.update" + // AuditActionPollingFailureAlertHandled 表示处理卡轮询失败告警。 + AuditActionPollingFailureAlertHandled = "polling_failure_alert.handle" // AuditActionPollingConcurrencyUpdated 表示更新轮询任务并发配置。 AuditActionPollingConcurrencyUpdated = "polling_concurrency.update" // AuditActionPollingConcurrencyReset 表示人工重置轮询并发计数。 @@ -781,6 +789,12 @@ const ( AuditResourceRateLimitGroup = "rate_limit_group" // AuditResourceGatewayErrorRule 表示上游错误分类规则资源。 AuditResourceGatewayErrorRule = "gateway_error_rule" + // AuditResourceCarrierPollingParam 表示运营商账户调度参数资源。 + AuditResourceCarrierPollingParam = "carrier_polling_param" + // AuditResourcePollingCheckScene 表示请求检查场景资源。 + AuditResourcePollingCheckScene = "polling_check_scene" + // AuditResourcePollingFailureAlert 表示卡轮询失败告警资源。 + AuditResourcePollingFailureAlert = "polling_failure_alert" // AuditResourcePollingConcurrencyConfig 表示轮询并发配置资源。 AuditResourcePollingConcurrencyConfig = "polling_concurrency" // AuditResourcePollingAlertRule 表示轮询告警规则资源。