ef5274864f
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 9m24s
资产解析返回的卡类型为 card，恢复轮询按 iot_card 匹配，导致改有效期后关闭轮询的卡不恢复。 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
304 lines
13 KiB
Go
304 lines
13 KiB
Go
package polling
|
||
|
||
import (
|
||
"context"
|
||
stderrors "errors"
|
||
"github.com/break/junhong_cmp_fiber/pkg/auditcontext"
|
||
"time"
|
||
|
||
"go.uber.org/zap"
|
||
"gorm.io/gorm"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/infrastructure/pollingpause"
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/internal/model/dto"
|
||
pausequery "github.com/break/junhong_cmp_fiber/internal/query/pollingpause"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||
"github.com/break/junhong_cmp_fiber/pkg/middleware"
|
||
)
|
||
|
||
// 勾选批量开关轮询、告警页关闭轮询与不轮询清单查询（add-polling-pause-list-and-batch-toggle）。
|
||
// 每项独立事务（经 UpdatePollingStatus 同事务写 enable_polling、清单与审计，提交后按生命周期移入或移出轮询），部分失败不影响其他项。
|
||
|
||
// BatchUpdatePollingStatus 勾选批量开关轮询：单次最多 500、校验数据权限、逐项处理并返回结果；批量开启跳过风险停机与已销户的卡。
|
||
func (s *AssetPollingService) BatchUpdatePollingStatus(ctx context.Context, assetType string, req *dto.BatchUpdatePollingStatusRequest) (*dto.BatchUpdatePollingStatusResponse, error) {
|
||
if middleware.GetUserTypeFromContext(ctx) == constants.UserTypeEnterprise {
|
||
return nil, errors.New(errors.CodeForbidden, "企业账号无权修改轮询开关")
|
||
}
|
||
if req == nil || req.EnablePolling == nil || len(req.AssetIDs) == 0 || len(req.AssetIDs) > constants.PollingBatchMaxSize {
|
||
return nil, errors.New(errors.CodeInvalidParam, "资产数量必须为 1～500")
|
||
}
|
||
seen := make(map[uint]bool, len(req.AssetIDs))
|
||
for _, id := range req.AssetIDs {
|
||
if id == 0 || seen[id] {
|
||
return nil, errors.New(errors.CodeInvalidParam, "资产ID不能为空或重复")
|
||
}
|
||
seen[id] = true
|
||
}
|
||
ctx = pollingpause.WithOrigin(ctx, pollingpause.Origin{Source: constants.PollingToggleSourceBatch, Reason: req.Reason})
|
||
resp := &dto.BatchUpdatePollingStatusResponse{Items: make([]dto.PollingToggleItem, 0, len(req.AssetIDs)), Impact: impactFor(*req.EnablePolling)}
|
||
for _, id := range req.AssetIDs {
|
||
item := s.ToggleAsset(ctx, assetType, id, *req.EnablePolling, true)
|
||
countToggle(resp, item)
|
||
resp.Items = append(resp.Items, item)
|
||
}
|
||
return resp, nil
|
||
}
|
||
|
||
// ToggleAsset 改一个资产的轮询开关并返回逐项结果；scoped 为 true 时按当前账号数据权限查找资产。
|
||
// 来源与原因由 ctx 中的 pollingpause.Origin 决定。
|
||
func (s *AssetPollingService) ToggleAsset(ctx context.Context, assetType string, assetID uint, enable, scoped bool) dto.PollingToggleItem {
|
||
item := dto.PollingToggleItem{AssetID: assetID}
|
||
current, identifier, extend, err := s.loadToggleTarget(ctx, assetType, assetID, scoped)
|
||
item.Identifier = identifier
|
||
if err != nil {
|
||
if stderrors.Is(err, gorm.ErrRecordNotFound) {
|
||
return toggleResult(item, constants.PollingToggleItemForbidden, "无权限操作该资产或资产不存在")
|
||
}
|
||
return toggleResult(item, constants.PollingToggleItemFailed, "查询资产失败")
|
||
}
|
||
if current == enable {
|
||
return toggleResult(item, constants.PollingToggleItemSkipped, "已是目标状态")
|
||
}
|
||
if enable && assetType == constants.AssetTypeIotCard && pollingpause.IsRiskExtend(extend) {
|
||
return toggleResult(item, constants.PollingToggleItemSkipped, pollingpause.RiskEnableDeniedReason(extend))
|
||
}
|
||
if err := s.UpdatePollingStatus(ctx, assetType, assetID, enable); err != nil {
|
||
message := "修改轮询开关失败"
|
||
var appErr *errors.AppError
|
||
if stderrors.As(err, &appErr) {
|
||
message = appErr.Error()
|
||
}
|
||
s.logger.Warn("修改资产轮询开关失败", zap.String("asset_type", assetType), zap.Uint("asset_id", assetID), zap.Error(err))
|
||
return toggleResult(item, constants.PollingToggleItemFailed, message)
|
||
}
|
||
return toggleResult(item, constants.PollingToggleItemSuccess, "")
|
||
}
|
||
|
||
func (s *AssetPollingService) loadToggleTarget(ctx context.Context, assetType string, assetID uint, scoped bool) (bool, string, string, error) {
|
||
query := s.db.WithContext(ctx)
|
||
switch assetType {
|
||
case constants.AssetTypeIotCard:
|
||
query = query.Model(&model.IotCard{})
|
||
if scoped {
|
||
query = middleware.ApplyStrictShopFilter(ctx, query)
|
||
}
|
||
var card model.IotCard
|
||
if err := query.Select("id", "iccid", "enable_polling", "gateway_extend").Where("id = ?", assetID).First(&card).Error; err != nil {
|
||
return false, "", "", err
|
||
}
|
||
return card.EnablePolling, card.ICCID, card.GatewayExtend, nil
|
||
case constants.AssetTypeDevice:
|
||
query = query.Model(&model.Device{})
|
||
if scoped {
|
||
query = middleware.ApplyStrictShopFilter(ctx, query)
|
||
}
|
||
var device model.Device
|
||
if err := query.Select("id", "imei", "virtual_no", "enable_polling").Where("id = ?", assetID).First(&device).Error; err != nil {
|
||
return false, "", "", err
|
||
}
|
||
identifier := device.IMEI
|
||
if identifier == "" {
|
||
identifier = device.VirtualNo
|
||
}
|
||
return device.EnablePolling, identifier, "", nil
|
||
default:
|
||
return false, "", "", errors.New(errors.CodeInvalidParam, "资产类型无效")
|
||
}
|
||
}
|
||
|
||
// ClosePollingByAlerts 告警页勾选批量关闭轮询：关闭告警对应卡的轮询（来源"停机被拒告警处理"），成功或已关闭的告警改为"已关闭轮询"。
|
||
func (s *AssetPollingService) ClosePollingByAlerts(ctx context.Context, req *dto.ClosePollingByAlertsRequest) (*dto.BatchUpdatePollingStatusResponse, error) {
|
||
if err := pausequery.EnsureAllowed(middleware.GetUserTypeFromContext(ctx)); err != nil {
|
||
return nil, err
|
||
}
|
||
if req == nil || len(req.AlertIDs) == 0 || len(req.AlertIDs) > constants.PollingBatchMaxSize {
|
||
return nil, errors.New(errors.CodeInvalidParam, "告警数量必须为 1～500")
|
||
}
|
||
var alerts []model.PollingFailureAlert
|
||
if err := s.db.WithContext(ctx).Where("id IN ? AND alert_level = ? AND card_id > 0", req.AlertIDs, constants.PollingAlertLevelCard).
|
||
Find(&alerts).Error; err != nil {
|
||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询失败告警失败")
|
||
}
|
||
operatorID := middleware.GetUserIDFromContext(ctx)
|
||
ctx = pollingpause.WithOrigin(ctx, pollingpause.Origin{Source: constants.PollingToggleSourceAlert, Reason: req.Reason})
|
||
resp := &dto.BatchUpdatePollingStatusResponse{Items: []dto.PollingToggleItem{}, Impact: impactFor(false)}
|
||
done := map[uint]dto.PollingToggleItem{}
|
||
found := map[uint]bool{}
|
||
for _, alert := range alerts {
|
||
found[alert.ID] = true
|
||
item, ok := done[alert.CardID]
|
||
if !ok {
|
||
item = s.ToggleAsset(ctx, constants.AssetTypeIotCard, alert.CardID, false, false)
|
||
done[alert.CardID] = item
|
||
countToggle(resp, item)
|
||
resp.Items = append(resp.Items, item)
|
||
}
|
||
if item.Status != constants.PollingToggleItemSuccess && item.Status != constants.PollingToggleItemSkipped {
|
||
continue
|
||
}
|
||
now := time.Now()
|
||
if err := s.db.WithContext(ctx).Model(&model.PollingFailureAlert{}).Where("id = ?", alert.ID).Updates(map[string]any{
|
||
"status": constants.PollingAlertStatusPollingDisabled, "handled_by": operatorID, "handled_at": now,
|
||
"handle_remark": req.Reason, "updated_at": now,
|
||
}).Error; err != nil {
|
||
s.logger.Warn("更新告警为已关闭轮询失败", zap.Uint("alert_id", alert.ID), zap.Error(err))
|
||
}
|
||
}
|
||
for _, id := range req.AlertIDs {
|
||
if !found[id] {
|
||
item := toggleResult(dto.PollingToggleItem{AssetID: 0}, constants.PollingToggleItemForbidden, "告警不存在或不是卡级告警")
|
||
countToggle(resp, item)
|
||
resp.Items = append(resp.Items, item)
|
||
}
|
||
}
|
||
return resp, nil
|
||
}
|
||
|
||
// ListPauses 查询不轮询清单（仅超级管理员与平台账号）。
|
||
func (s *AssetPollingService) ListPauses(ctx context.Context, req *dto.CardPollingPauseListRequest) (*dto.CardPollingPausePageResult, error) {
|
||
if err := pausequery.EnsureAllowed(middleware.GetUserTypeFromContext(ctx)); err != nil {
|
||
return nil, err
|
||
}
|
||
filter, err := PauseFilter(req.AssetType, req.CarrierID, req.Source, req.Status, req.Identifier, req.OperatorID, req.StartTime, req.EndTime)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
page, size := req.Page, req.PageSize
|
||
if page <= 0 {
|
||
page = constants.DefaultPage
|
||
}
|
||
if size <= 0 {
|
||
size = constants.DefaultPageSize
|
||
}
|
||
if size > constants.MaxPageSize {
|
||
size = constants.MaxPageSize
|
||
}
|
||
query := pausequery.NewQuery(s.db)
|
||
total, err := query.Count(ctx, filter)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
rows, err := query.Page(ctx, filter, (page-1)*size, size)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
items := make([]dto.CardPollingPauseItem, 0, len(rows))
|
||
for index := range rows {
|
||
items = append(items, ToPauseItem(&rows[index]))
|
||
}
|
||
return &dto.CardPollingPausePageResult{Items: items, Total: total, Page: page, PageSize: size}, nil
|
||
}
|
||
|
||
// Impact 返回开关轮询的影响说明。
|
||
func (s *AssetPollingService) Impact() *dto.PollingToggleImpactResponse {
|
||
return &dto.PollingToggleImpactResponse{Disable: impactFor(false), Enable: impactFor(true)}
|
||
}
|
||
|
||
// PauseFilter 由请求参数构造不轮询清单筛选条件（列表与导出共用）。
|
||
func PauseFilter(assetType string, carrierID *uint, source string, status *int, identifier string, operatorID *uint, start, end string) (pausequery.Filter, error) {
|
||
filter := pausequery.Filter{AssetType: assetType, CarrierID: carrierID, Source: source, Status: status, Identifier: identifier, OperatorID: operatorID}
|
||
for _, item := range []struct {
|
||
raw string
|
||
target **time.Time
|
||
}{{start, &filter.StartTime}, {end, &filter.EndTime}} {
|
||
if item.raw == "" {
|
||
continue
|
||
}
|
||
parsed, err := time.Parse(time.RFC3339, item.raw)
|
||
if err != nil {
|
||
return filter, errors.New(errors.CodeInvalidParam, "时间格式必须为带时区的 RFC3339")
|
||
}
|
||
*item.target = &parsed
|
||
}
|
||
return filter, nil
|
||
}
|
||
|
||
// ToPauseItem 把清单查询行转换为接口返回项。
|
||
func ToPauseItem(row *pausequery.Row) dto.CardPollingPauseItem {
|
||
return dto.CardPollingPauseItem{
|
||
ID: row.ID, AssetType: row.AssetType, AssetID: row.AssetID, Identifier: row.Identifier, CarrierID: row.CarrierID,
|
||
CarrierName: row.CarrierName, EnablePolling: row.EnablePolling, GatewayExtend: row.GatewayExtend,
|
||
Status: row.Status, StatusName: constants.PollingPauseStatusName(row.Status),
|
||
CloseSource: row.CloseSource, CloseSourceName: constants.PollingToggleSourceName(row.CloseSource), CloseReason: row.CloseReason,
|
||
ClosedBy: row.ClosedBy, ClosedByName: row.ClosedByName, ClosedAt: row.ClosedAt, ImportTaskID: row.ImportTaskID,
|
||
RestoreSource: row.RestoreSource, RestoreSourceName: constants.PollingToggleSourceName(row.RestoreSource), RestoreReason: row.RestoreReason,
|
||
RestoredBy: row.RestoredBy, RestoredByName: row.RestoredByName, RestoredAt: row.RestoredAt,
|
||
}
|
||
}
|
||
|
||
func impactFor(enable bool) []string {
|
||
source := constants.PollingToggleImpactDisable
|
||
if enable {
|
||
source = constants.PollingToggleImpactEnable
|
||
}
|
||
return append([]string(nil), source...)
|
||
}
|
||
|
||
func toggleResult(item dto.PollingToggleItem, status int, reason string) dto.PollingToggleItem {
|
||
item.Status, item.StatusName, item.Reason = status, constants.PollingToggleItemStatusName(status), reason
|
||
return item
|
||
}
|
||
|
||
func countToggle(resp *dto.BatchUpdatePollingStatusResponse, item dto.PollingToggleItem) {
|
||
switch item.Status {
|
||
case constants.PollingToggleItemSuccess:
|
||
resp.SuccessCount++
|
||
case constants.PollingToggleItemSkipped:
|
||
resp.SkipCount++
|
||
default:
|
||
resp.FailCount++
|
||
}
|
||
}
|
||
|
||
// RestoreAssetPolling 套餐生效类入口自动恢复轮询（订单套餐生效、改有效期后有生效套餐）：
|
||
// 卡关闭轮询且非风险停机、已销户时开启；设备关闭轮询时开启设备（级联开启设备下非风险停机、非已销户的卡），
|
||
// 设备已开启时开启设备下单独关闭的卡。返回是否恢复了任何资产。不做数据权限过滤（系统入口）。
|
||
func (s *AssetPollingService) RestoreAssetPolling(ctx context.Context, assetType string, assetID uint, source string) bool {
|
||
if s == nil || assetID == 0 {
|
||
return false
|
||
}
|
||
ctx = pollingpause.WithOrigin(ctx, pollingpause.Origin{Source: source, Reason: constants.PollingToggleSourceName(source) + "，自动恢复轮询"})
|
||
if auditcontext.From(ctx).ActorKind == "" {
|
||
ctx = auditcontext.With(ctx, auditcontext.Context{ActorKind: constants.AuditActorSystemTask, ActorID: "polling_auto_restore",
|
||
ActorName: "套餐生效自动恢复轮询", Source: constants.AuditSourceWorker})
|
||
}
|
||
if assetType == constants.AssetResolveTypeCard {
|
||
assetType = constants.AssetTypeIotCard
|
||
}
|
||
restored := false
|
||
switch assetType {
|
||
case constants.AssetTypeIotCard:
|
||
restored = s.restoreCard(ctx, assetID)
|
||
case constants.AssetTypeDevice:
|
||
item := s.ToggleAsset(ctx, constants.AssetTypeDevice, assetID, true, false)
|
||
restored = item.Status == constants.PollingToggleItemSuccess
|
||
if item.Status == constants.PollingToggleItemSkipped {
|
||
bindings, err := s.deviceBindingStore.ListByDeviceID(ctx, assetID)
|
||
if err != nil {
|
||
s.logger.Warn("自动恢复轮询查询设备绑定卡失败", zap.Uint("device_id", assetID), zap.Error(err))
|
||
return false
|
||
}
|
||
for _, binding := range bindings {
|
||
if s.restoreCard(ctx, binding.IotCardID) {
|
||
restored = true
|
||
}
|
||
}
|
||
}
|
||
}
|
||
if restored {
|
||
s.logger.Info("套餐生效后自动恢复轮询", zap.String("asset_type", assetType), zap.Uint("asset_id", assetID), zap.String("source", source))
|
||
}
|
||
return restored
|
||
}
|
||
|
||
// restoreCard 恢复单张卡的轮询；已开启、风险停机或已销户时不处理。
|
||
func (s *AssetPollingService) restoreCard(ctx context.Context, cardID uint) bool {
|
||
item := s.ToggleAsset(ctx, constants.AssetTypeIotCard, cardID, true, false)
|
||
if item.Status == constants.PollingToggleItemFailed {
|
||
s.logger.Warn("自动恢复卡轮询失败", zap.Uint("card_id", cardID), zap.String("reason", item.Reason))
|
||
}
|
||
return item.Status == constants.PollingToggleItemSuccess
|
||
}
|