feat: 下线设备保护期
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 9m30s
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 9m30s
- 后台设备停复机不再设置 1 小时保护期,也不再因保护期拒绝相反操作;单卡手动停复机去掉保护期检查 - 删除保护期一致性轮询(polling:protect)及其强制停复机方法;调度、初始化、监控、匹配预览不再包含保护期;已入队的旧任务由空处理函数直接确认,下一次发布移除 - 轮询配置接口不再接受与返回 protect_check_interval;并发配置不再返回 protect,查询、更新、重置该类型时拒绝 - 资产详情 device_protect_status 固定返回 none 并标记废弃,兼容老版本客户端 - 删除保护期 Redis key 与时长常量;停机原因 protect_period 保留供历史数据与开放接口映射 - 迁移 000249 只把 protect_check_interval、last_protect_check_at 列注释标记为已废弃,不删列不改数据
This commit is contained in:
@@ -112,7 +112,7 @@ func (h *PollingPackageHandler) Handle(ctx context.Context, t *asynq.Task) error
|
||||
zap.String("stop_reason", freshCard.StopReason))
|
||||
}
|
||||
// 标记本次评估来自普通套餐轮询:无有效套餐场景据此才允许追加优先轮询请求,
|
||||
// 其余调用链(观测消费者、手动实名、保护期轮询等)不打标即不写。
|
||||
// 其余调用链(观测消费者、手动实名等)不打标即不写。
|
||||
evaluationCtx := priorityapp.WithPollingPriorityTrigger(ctx)
|
||||
if evalErr := h.stopResumeSvc.EvaluateAndAct(evaluationCtx, freshCard); evalErr != nil {
|
||||
h.base.logger.Warn("套餐检查后停复机评估失败",
|
||||
|
||||
@@ -1,166 +0,0 @@
|
||||
package task
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/hibiken/asynq"
|
||||
"go.uber.org/zap"
|
||||
"gorm.io/gorm"
|
||||
|
||||
cardObservationApp "github.com/break/junhong_cmp_fiber/internal/application/cardobservation"
|
||||
"github.com/break/junhong_cmp_fiber/internal/gateway"
|
||||
iot_card_svc "github.com/break/junhong_cmp_fiber/internal/service/iot_card"
|
||||
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||||
)
|
||||
|
||||
// PollingProtectHandler 保护期一致性检查任务处理器
|
||||
// 保护期内:直接调 Gateway 强制停/复机(绕过 EvaluateAndAct 三条件判断)
|
||||
// 保护期结束:调 EvaluateAndAct 重新评估正常停复机条件
|
||||
// 两种路径不可混淆:保护期内=强制修正;保护期结束=重新评估
|
||||
type PollingProtectHandler struct {
|
||||
base *PollingBase
|
||||
iotCardStore *postgres.IotCardStore
|
||||
deviceSimBindingStore *postgres.DeviceSimBindingStore
|
||||
stopResumeSvc iot_card_svc.StopResumeServiceInterface
|
||||
}
|
||||
|
||||
// NewPollingProtectHandler 创建保护期一致性检查任务处理器
|
||||
func NewPollingProtectHandler(
|
||||
db *gorm.DB,
|
||||
observationSeriesEvents cardObservationApp.SeriesEventWriter,
|
||||
base *PollingBase,
|
||||
gw *gateway.Client,
|
||||
iotCardStore *postgres.IotCardStore,
|
||||
deviceSimBindingStore *postgres.DeviceSimBindingStore,
|
||||
stopResumeSvc iot_card_svc.StopResumeServiceInterface,
|
||||
) *PollingProtectHandler {
|
||||
return &PollingProtectHandler{
|
||||
base: base,
|
||||
iotCardStore: iotCardStore,
|
||||
deviceSimBindingStore: deviceSimBindingStore,
|
||||
stopResumeSvc: stopResumeSvc,
|
||||
}
|
||||
}
|
||||
|
||||
// Handle 处理保护期一致性检查任务
|
||||
func (h *PollingProtectHandler) Handle(ctx context.Context, t *asynq.Task) error {
|
||||
startTime := time.Now()
|
||||
|
||||
cardID, ok := parseTaskPayload(t.Payload(), h.base.logger)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
ctx = withPollingWorkerAuditContext(ctx, constants.TaskTypePollingProtect, "保护期一致性轮询任务", t.ResultWriter().TaskID())
|
||||
|
||||
if !h.base.acquireConcurrency(ctx, constants.TaskTypePollingProtect) {
|
||||
h.base.logger.Debug("并发已满,重新入队", zap.Uint("card_id", cardID))
|
||||
return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingProtect)
|
||||
}
|
||||
defer h.base.releaseConcurrency(ctx, constants.TaskTypePollingProtect)
|
||||
|
||||
card, err := h.iotCardStore.GetByID(ctx, cardID)
|
||||
if err != nil {
|
||||
if isNotFound(err) {
|
||||
return nil
|
||||
}
|
||||
h.base.logger.Error("查询卡信息失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||||
return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingProtect)
|
||||
}
|
||||
|
||||
// 独立卡(未绑设备)跳过保护期检查
|
||||
if card.IsStandalone {
|
||||
return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingProtect)
|
||||
}
|
||||
|
||||
// 实名不满足的卡跳过保护期检查:停复机新规则总开关关闭时未实名即跳过,开启时只在通道开启实名管控且未实名时跳过
|
||||
if h.stopResumeSvc != nil {
|
||||
skip, skipErr := h.stopResumeSvc.ProtectCheckSkipsForRealname(ctx, card)
|
||||
if skipErr != nil {
|
||||
h.base.logger.Warn("保护期检查判断实名管控失败,本次跳过", zap.Uint("card_id", cardID), zap.Error(skipErr))
|
||||
return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingProtect)
|
||||
}
|
||||
if skip {
|
||||
return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingProtect)
|
||||
}
|
||||
} else if card.RealNameStatus != constants.RealNameStatusVerified {
|
||||
return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingProtect)
|
||||
}
|
||||
|
||||
binding, err := h.deviceSimBindingStore.GetActiveBindingByCardID(ctx, card.ID)
|
||||
if err != nil || binding == nil {
|
||||
return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingProtect)
|
||||
}
|
||||
deviceID := binding.DeviceID
|
||||
|
||||
stopProtectGeneration := h.base.redis.Get(ctx, constants.RedisDeviceProtectKey(deviceID, "stop")).Val()
|
||||
startProtectGeneration := h.base.redis.Get(ctx, constants.RedisDeviceProtectKey(deviceID, "start")).Val()
|
||||
stopProtect := stopProtectGeneration != ""
|
||||
startProtect := startProtectGeneration != ""
|
||||
|
||||
actionTaken := "no_action"
|
||||
|
||||
switch {
|
||||
case stopProtect && card.NetworkStatus == constants.NetworkStatusOnline:
|
||||
// 保护期内:停机保护期发现开机卡 → 强制停机(绕过 EvaluateAndAct)
|
||||
h.base.logger.Info("保护期一致性:停机保护期内发现开机卡,强制停机",
|
||||
zap.Uint("card_id", card.ID), zap.Uint("device_id", deviceID))
|
||||
if h.stopResumeSvc == nil {
|
||||
break
|
||||
}
|
||||
if err := h.stopResumeSvc.ForceStopCard(ctx, card, constants.StopReasonProtectPeriod); err != nil {
|
||||
h.base.logger.Error("保护期强制停机失败",
|
||||
zap.Uint("card_id", card.ID), zap.Error(err))
|
||||
h.base.updateStats(ctx, constants.TaskTypePollingProtect, false, time.Since(startTime))
|
||||
return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingProtect)
|
||||
}
|
||||
h.base.updateCardCache(ctx, cardID, map[string]any{"network_status": constants.NetworkStatusOffline})
|
||||
actionTaken = "forced_stop"
|
||||
|
||||
case startProtect && card.NetworkStatus == constants.NetworkStatusOffline:
|
||||
// 保护期内:复机保护期发现停机卡 → 强制复机(绕过 EvaluateAndAct)
|
||||
h.base.logger.Info("保护期一致性:复机保护期内发现停机卡,强制复机",
|
||||
zap.Uint("card_id", card.ID), zap.Uint("device_id", deviceID))
|
||||
if h.stopResumeSvc == nil {
|
||||
break
|
||||
}
|
||||
if err := h.stopResumeSvc.ForceStartCard(ctx, card); err != nil {
|
||||
h.base.logger.Error("保护期强制复机失败",
|
||||
zap.Uint("card_id", card.ID), zap.Error(err))
|
||||
h.base.updateStats(ctx, constants.TaskTypePollingProtect, false, time.Since(startTime))
|
||||
return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingProtect)
|
||||
}
|
||||
h.base.updateCardCache(ctx, cardID, map[string]any{"network_status": constants.NetworkStatusOnline})
|
||||
actionTaken = "forced_resume"
|
||||
|
||||
case !stopProtect && !startProtect:
|
||||
// 保护期结束:调 EvaluateAndAct 重新评估正常停复机条件
|
||||
if h.stopResumeSvc != nil {
|
||||
if evalErr := h.stopResumeSvc.EvaluateAndAct(ctx, card); evalErr != nil {
|
||||
h.base.logger.Warn("保护期结束后停复机评估失败",
|
||||
zap.Uint("card_id", cardID), zap.Error(evalErr))
|
||||
}
|
||||
}
|
||||
actionTaken = "evaluate"
|
||||
}
|
||||
|
||||
if h.base.verboseLog {
|
||||
h.base.logger.Info("保护期检查详情",
|
||||
zap.Uint("card_id", cardID),
|
||||
zap.String("iccid", card.ICCID),
|
||||
zap.Bool("stop_protect", stopProtect),
|
||||
zap.Bool("start_protect", startProtect),
|
||||
zap.Int("network_status_at_check", card.NetworkStatus),
|
||||
zap.String("action_taken", actionTaken))
|
||||
}
|
||||
|
||||
if updateErr := h.iotCardStore.UpdateFields(ctx, cardID, map[string]any{
|
||||
"last_protect_check_at": time.Now(),
|
||||
}); updateErr != nil {
|
||||
h.base.logger.Warn("更新保护期检查时间失败", zap.Uint("card_id", cardID), zap.Error(updateErr))
|
||||
}
|
||||
|
||||
h.base.updateStats(ctx, constants.TaskTypePollingProtect, true, time.Since(startTime))
|
||||
return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingProtect)
|
||||
}
|
||||
Reference in New Issue
Block a user