925 lines
36 KiB
Go
925 lines
36 KiB
Go
package iot_card
|
||
|
||
import (
|
||
"context"
|
||
stderrors "errors"
|
||
"strings"
|
||
"time"
|
||
|
||
cardObservationApp "github.com/break/junhong_cmp_fiber/internal/application/cardobservation"
|
||
"github.com/google/uuid"
|
||
"github.com/redis/go-redis/v9"
|
||
"go.uber.org/zap"
|
||
"gorm.io/gorm"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/gateway"
|
||
"github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
|
||
"github.com/break/junhong_cmp_fiber/internal/infrastructure/integrationlog"
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||
)
|
||
|
||
// StopResumeServiceInterface 停复机服务接口
|
||
// 供 4 个 Task Handler 通过接口注入，避免循环依赖
|
||
type StopResumeServiceInterface interface {
|
||
// EvaluateAndAct 停复机统一入口，根据卡的当前状态自动判断并执行停机或复机
|
||
EvaluateAndAct(ctx context.Context, card *model.IotCard) error
|
||
// ForceStopCard 强制停机单张卡，不执行正常停机条件判断。
|
||
ForceStopCard(ctx context.Context, card *model.IotCard, stopReason string) error
|
||
// ForceStartCard 强制复机单张卡，不执行正常复机条件判断。
|
||
ForceStartCard(ctx context.Context, card *model.IotCard) error
|
||
}
|
||
|
||
// 编译时验证 StopResumeService 实现了 StopResumeServiceInterface
|
||
var _ StopResumeServiceInterface = (*StopResumeService)(nil)
|
||
|
||
// StopResumeService 停复机服务
|
||
// 处理 IoT 卡的自动停机、复机和手动停复机逻辑
|
||
type StopResumeService struct {
|
||
db *gorm.DB
|
||
redis *redis.Client
|
||
iotCardStore *postgres.IotCardStore
|
||
packageUsageStore *postgres.PackageUsageStore
|
||
deviceSimBindingStore *postgres.DeviceSimBindingStore
|
||
gatewayClient *gateway.Client
|
||
logger *zap.Logger
|
||
pollingCallback PollingCallback
|
||
observationSeriesEvents cardObservationApp.SeriesEventWriter
|
||
auditWriter *audit.Writer
|
||
integration *integrationlog.Repository
|
||
|
||
maxRetries int
|
||
retryInterval time.Duration
|
||
}
|
||
|
||
// SetObservationSeriesEventWriter 注入停复机成功观测序列 Outbox Writer。
|
||
func (s *StopResumeService) SetObservationSeriesEventWriter(db *gorm.DB, writer cardObservationApp.SeriesEventWriter) {
|
||
s.db = db
|
||
s.observationSeriesEvents = writer
|
||
}
|
||
|
||
// SetUnifiedAudit 注入停复机统一审计和外部交互日志接缝。
|
||
func (s *StopResumeService) SetUnifiedAudit(writer *audit.Writer, integration *integrationlog.Repository) {
|
||
s.auditWriter = writer
|
||
s.integration = integration
|
||
}
|
||
|
||
// NewStopResumeService 创建停复机服务
|
||
func NewStopResumeService(
|
||
redis *redis.Client,
|
||
iotCardStore *postgres.IotCardStore,
|
||
packageUsageStore *postgres.PackageUsageStore,
|
||
deviceSimBindingStore *postgres.DeviceSimBindingStore,
|
||
gatewayClient *gateway.Client,
|
||
logger *zap.Logger,
|
||
) *StopResumeService {
|
||
return &StopResumeService{
|
||
redis: redis,
|
||
iotCardStore: iotCardStore,
|
||
packageUsageStore: packageUsageStore,
|
||
deviceSimBindingStore: deviceSimBindingStore,
|
||
gatewayClient: gatewayClient,
|
||
logger: logger,
|
||
maxRetries: 3,
|
||
retryInterval: 2 * time.Second,
|
||
}
|
||
}
|
||
|
||
// SetPollingCallback 设置轮询生命周期回调。
|
||
// 停复机成功改变卡状态后，需要立即按新状态重排轮询队列，避免继续使用旧 score。
|
||
func (s *StopResumeService) SetPollingCallback(callback PollingCallback) {
|
||
s.pollingCallback = callback
|
||
}
|
||
|
||
// EvaluateAndAct 停复机统一入口
|
||
// 根据卡的当前网络状态，自动判断并执行停机或复机操作
|
||
// 设备/单卡维度通过 card.IsStandalone 推导（false 则为设备维度，停/复机覆盖设备下所有卡）
|
||
func (s *StopResumeService) EvaluateAndAct(ctx context.Context, card *model.IotCard) error {
|
||
switch card.NetworkStatus {
|
||
case constants.NetworkStatusOnline:
|
||
// 卡在线，检查是否需要停机
|
||
reasons, err := s.checkStopReasons(ctx, card)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if len(reasons) == 0 {
|
||
return nil
|
||
}
|
||
// 取最高优先级原因（第一个）
|
||
primaryReason := reasons[0]
|
||
if !card.IsStandalone && isDeviceScopeReason(primaryReason) {
|
||
deviceID, bound, lookupErr := s.getCardDeviceID(ctx, card)
|
||
if lookupErr != nil {
|
||
s.logger.Error("查询设备绑定关系失败，降级为单卡停机",
|
||
zap.Uint("card_id", card.ID), zap.Error(lookupErr))
|
||
// 降级：查不到绑定时按单卡处理
|
||
} else if bound {
|
||
return s.stopDeviceCards(ctx, deviceID, primaryReason)
|
||
}
|
||
}
|
||
return s.stopCardWithRetry(ctx, card, primaryReason)
|
||
|
||
case constants.NetworkStatusOffline:
|
||
// 卡停机，检查是否可以复机
|
||
// 只处理轮询系统引起的停机原因（手动停机等不自动复机）
|
||
if !isPollingStopReason(card.StopReason) {
|
||
return nil
|
||
}
|
||
shouldResume, err := s.shouldResume(ctx, card)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !shouldResume {
|
||
return nil
|
||
}
|
||
if !card.IsStandalone {
|
||
deviceID, bound, lookupErr := s.getCardDeviceID(ctx, card)
|
||
if lookupErr != nil {
|
||
s.logger.Error("查询设备绑定关系失败，降级为单卡复机",
|
||
zap.Uint("card_id", card.ID), zap.Error(lookupErr))
|
||
} else if bound {
|
||
return s.resumeDeviceCards(ctx, deviceID)
|
||
}
|
||
}
|
||
return s.resumeSingleCard(ctx, card.ID)
|
||
|
||
default:
|
||
return nil
|
||
}
|
||
}
|
||
|
||
// isRiskGatewayExtend 判断网关扩展状态是否为风险状态（风险停机或已销户）
|
||
// 独立卡命中此状态时禁止发起复机
|
||
func isRiskGatewayExtend(extend string) bool {
|
||
trimmed := strings.TrimSpace(extend)
|
||
return trimmed == constants.GatewayCardExtendRiskStop ||
|
||
trimmed == constants.GatewayCardExtendCancelled
|
||
}
|
||
|
||
// isPollingStopReason 判断停机原因是否为轮询系统引起（可自动复机）
|
||
func isPollingStopReason(reason string) bool {
|
||
switch reason {
|
||
case constants.StopReasonTrafficExhausted,
|
||
constants.StopReasonNoPackage,
|
||
constants.StopReasonNotRealname:
|
||
return true
|
||
}
|
||
return false
|
||
}
|
||
|
||
// isDeviceScopeReason 判断停机原因是否应扩散到设备维度
|
||
// 套餐/流量是设备共享资源，需整体停机；实名是单卡属性，只停该卡
|
||
func isDeviceScopeReason(reason string) bool {
|
||
return reason == constants.StopReasonNoPackage ||
|
||
reason == constants.StopReasonTrafficExhausted
|
||
}
|
||
|
||
// getCardDeviceID 获取卡所属设备ID（非独立卡时有效）
|
||
// 返回：deviceID, isDeviceBound, error
|
||
func (s *StopResumeService) getCardDeviceID(ctx context.Context, card *model.IotCard) (uint, bool, error) {
|
||
if card.IsStandalone {
|
||
return 0, false, nil
|
||
}
|
||
binding, err := s.deviceSimBindingStore.GetActiveBindingByCardID(ctx, card.ID)
|
||
if err != nil {
|
||
if stderrors.Is(err, gorm.ErrRecordNotFound) {
|
||
return 0, false, nil
|
||
}
|
||
return 0, false, err
|
||
}
|
||
if binding == nil {
|
||
return 0, false, nil
|
||
}
|
||
return binding.DeviceID, true, nil
|
||
}
|
||
|
||
// hasValidPackage 检查卡是否有有效套餐（status IN 0,1）
|
||
// 修复Bug1：绑定设备的卡查 device_id，独立卡查 iot_card_id
|
||
func (s *StopResumeService) hasValidPackage(ctx context.Context, card *model.IotCard) (bool, error) {
|
||
if !card.IsStandalone {
|
||
deviceID, bound, lookupErr := s.getCardDeviceID(ctx, card)
|
||
if lookupErr != nil {
|
||
return false, lookupErr
|
||
}
|
||
if bound {
|
||
return s.packageUsageStore.HasValidByCarrier(ctx, "device", deviceID)
|
||
}
|
||
}
|
||
return s.packageUsageStore.HasValidByCarrier(ctx, "iot_card", card.ID)
|
||
}
|
||
|
||
// isTrafficExhausted 检查卡的流量是否已耗尽
|
||
// 条件：活跃套餐 status=2（已用完），或 data_usage_mb >= data_limit_mb（data_limit_mb > 0）
|
||
// 业务约定：data_limit_mb=0 表示无限流量套餐，永远不判定为耗尽
|
||
func (s *StopResumeService) isTrafficExhausted(ctx context.Context, card *model.IotCard) (bool, error) {
|
||
carrierType := "iot_card"
|
||
carrierID := card.ID
|
||
|
||
if !card.IsStandalone {
|
||
deviceID, bound, lookupErr := s.getCardDeviceID(ctx, card)
|
||
if lookupErr != nil {
|
||
return false, lookupErr
|
||
}
|
||
if bound {
|
||
carrierType = "device"
|
||
carrierID = deviceID
|
||
}
|
||
}
|
||
|
||
packages, err := s.packageUsageStore.ListActiveOrDepletedByCarrier(ctx, carrierType, carrierID)
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
if len(packages) == 0 {
|
||
return false, nil
|
||
}
|
||
|
||
for _, pkg := range packages {
|
||
if pkg == nil {
|
||
continue
|
||
}
|
||
if !pkg.IsTrafficDepleted() {
|
||
return false, nil
|
||
}
|
||
}
|
||
return true, nil
|
||
}
|
||
|
||
// isRealnameOK 检查卡是否满足实名要求
|
||
// 行业卡（card_category='industry'）无需实名；其他卡需 real_name_status=1
|
||
func (s *StopResumeService) isRealnameOK(card *model.IotCard) bool {
|
||
return card.CardCategory == constants.CardCategoryIndustry ||
|
||
card.RealNameStatus == constants.RealNameStatusVerified
|
||
}
|
||
|
||
// checkStopReasons 检查卡的停机原因列表，按优先级排序
|
||
// 优先级：no_package > traffic_exhausted > not_realname
|
||
func (s *StopResumeService) checkStopReasons(ctx context.Context, card *model.IotCard) ([]string, error) {
|
||
var reasons []string
|
||
|
||
// 条件A：流量耗尽
|
||
exhausted, err := s.isTrafficExhausted(ctx, card)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if exhausted {
|
||
reasons = append(reasons, constants.StopReasonTrafficExhausted)
|
||
}
|
||
|
||
// 条件B：无有效套餐（仅在未命中流量耗尽时判断）
|
||
hasPackage, err := s.hasValidPackage(ctx, card)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if !hasPackage && !exhausted {
|
||
reasons = append(reasons, constants.StopReasonNoPackage)
|
||
}
|
||
|
||
// 条件C：非行业卡且未实名
|
||
if !s.isRealnameOK(card) {
|
||
reasons = append(reasons, constants.StopReasonNotRealname)
|
||
}
|
||
|
||
return reasons, nil
|
||
}
|
||
|
||
// shouldResume 检查停机卡是否满足复机条件
|
||
// 须同时满足：有有效套餐 + 流量未耗尽 + 实名OK
|
||
func (s *StopResumeService) shouldResume(ctx context.Context, card *model.IotCard) (bool, error) {
|
||
hasPackage, err := s.hasValidPackage(ctx, card)
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
if !hasPackage {
|
||
return false, nil
|
||
}
|
||
|
||
exhausted, err := s.isTrafficExhausted(ctx, card)
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
if exhausted {
|
||
return false, nil
|
||
}
|
||
|
||
return s.isRealnameOK(card), nil
|
||
}
|
||
|
||
// stopDeviceCards 停机设备下所有在线卡（含设备维度幂等锁）
|
||
// 防止设备下多张卡并发触发 EvaluateAndAct 导致重复 Gateway 停机调用
|
||
func (s *StopResumeService) stopDeviceCards(ctx context.Context, deviceID uint, stopReason string) error {
|
||
// 设备维度幂等锁：防止多协程同时对同一设备停机
|
||
lockKey := constants.RedisPollingDeviceOpLockKey(deviceID)
|
||
locked, lockErr := s.redis.SetNX(ctx, lockKey, "1", 30*time.Second).Result()
|
||
if lockErr != nil {
|
||
return errors.Wrap(errors.CodeRedisError, lockErr, "获取设备停机锁失败")
|
||
}
|
||
if !locked {
|
||
s.logger.Debug("设备停机操作已在进行中，跳过重复调用",
|
||
zap.Uint("device_id", deviceID))
|
||
return nil
|
||
}
|
||
defer s.redis.Del(ctx, lockKey)
|
||
|
||
// 查询设备下所有在线卡
|
||
cards, err := s.iotCardStore.ListByDeviceIDAndNetworkStatus(ctx, deviceID, constants.NetworkStatusOnline)
|
||
if err != nil {
|
||
s.logger.Error("查询设备在线卡失败",
|
||
zap.Uint("device_id", deviceID), zap.Error(err))
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "查询设备在线卡失败")
|
||
}
|
||
|
||
var cardErrors []error
|
||
for _, card := range cards {
|
||
if stopErr := s.stopCardWithRetry(ctx, card, stopReason); stopErr != nil {
|
||
cardErrors = append(cardErrors, stopErr)
|
||
s.logger.Warn("设备卡停机失败，继续处理其他卡",
|
||
zap.Uint("device_id", deviceID),
|
||
zap.Uint("card_id", card.ID),
|
||
zap.Error(stopErr))
|
||
}
|
||
}
|
||
if len(cardErrors) > 0 {
|
||
return errors.Wrap(errors.CodeInternalError, stderrors.Join(cardErrors...), "设备停机存在未成功处理的卡")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// resumeDeviceCards 复机设备下满足条件的停机卡（含设备维度幂等锁）
|
||
// 遍历时对每张卡检查实名状态：未实名普通卡更新 stop_reason='not_realname' 后跳过
|
||
func (s *StopResumeService) resumeDeviceCards(ctx context.Context, deviceID uint) error {
|
||
// 设备维度幂等锁：与 stopDeviceCards 共用，防止停/复机并发
|
||
lockKey := constants.RedisPollingDeviceOpLockKey(deviceID)
|
||
locked, lockErr := s.redis.SetNX(ctx, lockKey, "1", 30*time.Second).Result()
|
||
if lockErr != nil {
|
||
return errors.Wrap(errors.CodeRedisError, lockErr, "获取设备复机锁失败")
|
||
}
|
||
if !locked {
|
||
s.logger.Debug("设备复机操作已在进行中，跳过重复调用",
|
||
zap.Uint("device_id", deviceID))
|
||
return nil
|
||
}
|
||
defer s.redis.Del(ctx, lockKey)
|
||
|
||
// 查询设备下因轮询原因停机的卡
|
||
cards, err := s.iotCardStore.ListByDeviceIDAndPollingStopReasons(ctx, deviceID)
|
||
if err != nil {
|
||
s.logger.Error("查询设备停机卡失败",
|
||
zap.Uint("device_id", deviceID), zap.Error(err))
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "查询设备停机卡失败")
|
||
}
|
||
|
||
var cardErrors []error
|
||
for _, card := range cards {
|
||
if !s.isRealnameOK(card) {
|
||
if updateErr := s.updateCardStopReasonWithAudit(ctx, card, constants.StopReasonNotRealname); updateErr != nil {
|
||
cardErrors = append(cardErrors, updateErr)
|
||
s.logger.Warn("更新未实名卡停机原因失败",
|
||
zap.Uint("card_id", card.ID), zap.Error(updateErr))
|
||
}
|
||
continue
|
||
}
|
||
if resumeErr := s.resumeSingleCard(ctx, card.ID); resumeErr != nil {
|
||
cardErrors = append(cardErrors, resumeErr)
|
||
s.logger.Warn("设备卡复机失败，继续处理其他卡",
|
||
zap.Uint("device_id", deviceID),
|
||
zap.Uint("card_id", card.ID),
|
||
zap.Error(resumeErr))
|
||
}
|
||
}
|
||
if len(cardErrors) > 0 {
|
||
return errors.Wrap(errors.CodeInternalError, stderrors.Join(cardErrors...), "设备复机存在未成功处理的卡")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// CheckAndStopCard 检查流量耗尽并停机（旧入口，保留兼容性）
|
||
// 新代码应优先使用 EvaluateAndAct
|
||
func (s *StopResumeService) CheckAndStopCard(ctx context.Context, cardID uint) error {
|
||
card, err := s.iotCardStore.GetByID(ctx, cardID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return s.EvaluateAndAct(ctx, card)
|
||
}
|
||
|
||
// ResumeCardIfStopped 套餐激活或流量重置后自动复机入口
|
||
// 支持 iot_card 和 device 两种载体类型
|
||
func (s *StopResumeService) ResumeCardIfStopped(ctx context.Context, carrierType string, carrierID uint) error {
|
||
switch carrierType {
|
||
case "iot_card":
|
||
return s.resumeSingleCard(ctx, carrierID)
|
||
case "device":
|
||
return s.resumeDeviceCards(ctx, carrierID)
|
||
default:
|
||
return nil
|
||
}
|
||
}
|
||
|
||
// ForceStopCard 强制停机单张卡，不执行正常停机条件判断。
|
||
func (s *StopResumeService) ForceStopCard(ctx context.Context, card *model.IotCard, stopReason string) error {
|
||
if card == nil || card.ID == 0 {
|
||
return errors.New(errors.CodeInvalidParam)
|
||
}
|
||
return s.stopCardWithRetry(ctx, card, stopReason)
|
||
}
|
||
|
||
// ForceStartCard 强制复机单张卡，不执行正常复机条件判断。
|
||
func (s *StopResumeService) ForceStartCard(ctx context.Context, card *model.IotCard) error {
|
||
if card == nil || card.ID == 0 {
|
||
return errors.New(errors.CodeInvalidParam)
|
||
}
|
||
actionCode, summary := constants.AuditActionIotCardAutoStarted, "自动恢复 IoT 卡网络"
|
||
attempt, err := s.resumeCardWithRetry(ctx, card, actionCode, summary)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err := s.updateCardAndAppendNetworkSeries(ctx, card, map[string]any{
|
||
"network_status": constants.NetworkStatusOnline,
|
||
"resumed_at": time.Now(),
|
||
"stop_reason": "",
|
||
}, constants.CardObservationSceneBusinessResume, "online", uuid.NewString(), actionCode, summary, attempt.log.IntegrationID); err != nil {
|
||
if logErr := s.completeCardCommandAttempt(ctx, attempt, nil, false); logErr != nil {
|
||
s.logger.Error("终结保护期复机 Integration Log 失败", zap.String("integration_id", attempt.log.IntegrationID), zap.Error(logErr))
|
||
}
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"结果待核对", constants.AuditResultUnknown,
|
||
attempt.log.IntegrationID, cardSnapshot(card), map[string]any{"requested_network_status": constants.NetworkStatusOnline}, err)
|
||
return err
|
||
}
|
||
s.reschedulePolling(ctx, card.ID)
|
||
if err := s.completeCardCommandAttempt(ctx, attempt, nil, true); err != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "终结保护期复机 Integration Log 失败")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// resumeSingleCard 对单张卡执行复机逻辑
|
||
// 依次检查：已开机则跳过 → 非轮询停机原因则跳过 → 不满足复机条件则跳过 → 加锁 → 调 Gateway → 更新 DB
|
||
func (s *StopResumeService) resumeSingleCard(ctx context.Context, cardID uint) error {
|
||
card, err := s.iotCardStore.GetByID(ctx, cardID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
if card.NetworkStatus == constants.NetworkStatusOnline {
|
||
return nil
|
||
}
|
||
|
||
// 只处理轮询系统引起的停机
|
||
if !isPollingStopReason(card.StopReason) {
|
||
s.logger.Debug("卡非轮询停机原因，不自动复机",
|
||
zap.Uint("card_id", cardID),
|
||
zap.String("stop_reason", card.StopReason))
|
||
return nil
|
||
}
|
||
|
||
// 检查复机条件
|
||
resume, err := s.shouldResume(ctx, card)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !resume {
|
||
return nil
|
||
}
|
||
|
||
// 分布式锁防止同一张卡并发触发多次 Gateway 调用，TTL 30s
|
||
lockKey := constants.RedisCardResumeLockKey(cardID)
|
||
locked, lockErr := s.redis.SetNX(ctx, lockKey, "1", 30*time.Second).Result()
|
||
if lockErr != nil {
|
||
s.logger.Warn("获取复机分布式锁失败，跳过本次复机",
|
||
zap.Uint("card_id", cardID),
|
||
zap.Error(lockErr))
|
||
return nil
|
||
}
|
||
if !locked {
|
||
s.logger.Debug("复机操作进行中，跳过", zap.Uint("card_id", cardID))
|
||
return nil
|
||
}
|
||
defer s.redis.Del(ctx, lockKey)
|
||
|
||
actionCode, summary := constants.AuditActionIotCardAutoStarted, "自动恢复 IoT 卡网络"
|
||
attempt, err := s.resumeCardWithRetry(ctx, card, actionCode, summary)
|
||
if err != nil {
|
||
s.logger.Error("调用运营商复机接口失败",
|
||
zap.Uint("card_id", cardID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Error(err))
|
||
return err
|
||
}
|
||
|
||
now := time.Now()
|
||
if err := s.updateCardAndAppendNetworkSeries(ctx, card, map[string]any{
|
||
"network_status": constants.NetworkStatusOnline,
|
||
"resumed_at": now,
|
||
"stop_reason": "",
|
||
}, constants.CardObservationSceneBusinessResume, "online", uuid.NewString(), actionCode, summary, attempt.log.IntegrationID); err != nil {
|
||
if logErr := s.completeCardCommandAttempt(ctx, attempt, nil, false); logErr != nil {
|
||
s.logger.Error("终结复机 Integration Log 失败", zap.String("integration_id", attempt.log.IntegrationID), zap.Error(logErr))
|
||
}
|
||
s.logger.Error("复机 Gateway 成功但 DB 更新失败",
|
||
zap.Uint("card_id", cardID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Error(err))
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"结果待核对", constants.AuditResultUnknown,
|
||
attempt.log.IntegrationID,
|
||
map[string]any{"network_status": card.NetworkStatus, "stop_reason": card.StopReason},
|
||
map[string]any{"requested_network_status": constants.NetworkStatusOnline}, err)
|
||
return err
|
||
}
|
||
s.reschedulePolling(ctx, card.ID)
|
||
if logErr := s.completeCardCommandAttempt(ctx, attempt, nil, true); logErr != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, logErr, "终结复机 Integration Log 失败")
|
||
}
|
||
|
||
s.logger.Info("卡已自动复机",
|
||
zap.Uint("card_id", cardID),
|
||
zap.String("iccid", card.ICCID))
|
||
|
||
return nil
|
||
}
|
||
|
||
// stopCardWithRetry 调用运营商停机接口（带重试机制），并更新 DB 停机原因
|
||
func (s *StopResumeService) stopCardWithRetry(ctx context.Context, card *model.IotCard, stopReason string) error {
|
||
actionCode, summary := stopAuditAction(ctx, stopReason)
|
||
if s.gatewayClient == nil {
|
||
failErr := errors.New(errors.CodeInternalError, "Gateway 未配置，停复机操作不可用")
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"失败", constants.AuditResultFailed, "",
|
||
cardSnapshot(card), nil, failErr)
|
||
return failErr
|
||
}
|
||
|
||
s.logger.Info("调用网关停机",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.String("stop_reason", stopReason))
|
||
|
||
seriesKey := cardCommandSeriesKey(ctx)
|
||
attemptObserver := &cardCommandAttemptObserver{
|
||
service: s, card: card, operation: constants.IntegrationOperationGatewayStopCard,
|
||
scene: constants.CardObservationSceneBusinessStop, seriesKey: seriesKey,
|
||
}
|
||
gatewayCtx := gateway.WithAttemptObserver(ctx, attemptObserver)
|
||
var lastErr error
|
||
lastIntegrationID := ""
|
||
for i := 0; i < s.maxRetries; i++ {
|
||
if i > 0 {
|
||
s.logger.Debug("重试调用停机接口",
|
||
zap.Int("attempt", i+1),
|
||
zap.String("iccid", card.ICCID))
|
||
time.Sleep(s.retryInterval)
|
||
}
|
||
|
||
callErr := s.gatewayClient.StopCard(gatewayCtx, &gateway.CardOperationReq{CardNo: card.ICCID})
|
||
lastIntegrationID = attemptObserver.lastIntegrationID
|
||
if attemptObserver.recordingErr != nil {
|
||
s.logger.Error("记录停机 Integration Log 失败", zap.String("integration_id", lastIntegrationID), zap.Error(attemptObserver.recordingErr))
|
||
lastErr = attemptObserver.lastCallErr
|
||
if lastErr == nil {
|
||
lastErr = attemptObserver.recordingErr
|
||
}
|
||
break
|
||
}
|
||
if callErr == nil {
|
||
attempt := attemptObserver.successful
|
||
s.logger.Info("网关停机成功",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID))
|
||
|
||
now := time.Now()
|
||
if updateErr := s.updateCardAndAppendNetworkSeries(ctx, card, map[string]any{
|
||
"network_status": constants.NetworkStatusOffline,
|
||
"stopped_at": now,
|
||
"stop_reason": stopReason,
|
||
}, constants.CardObservationSceneBusinessStop, "offline", uuid.NewString(), actionCode, summary, lastIntegrationID); updateErr != nil {
|
||
if logErr := s.completeCardCommandAttempt(ctx, attempt, nil, false); logErr != nil {
|
||
s.logger.Error("终结停机 Integration Log 失败", zap.String("integration_id", lastIntegrationID), zap.Error(logErr))
|
||
}
|
||
s.logger.Error("停机 Gateway 成功但 DB 更新失败",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Error(updateErr))
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"结果待核对", constants.AuditResultUnknown,
|
||
lastIntegrationID,
|
||
map[string]any{"network_status": card.NetworkStatus, "stop_reason": card.StopReason},
|
||
map[string]any{"requested_network_status": constants.NetworkStatusOffline, "stop_reason": stopReason}, updateErr)
|
||
return updateErr
|
||
}
|
||
|
||
s.reschedulePolling(ctx, card.ID)
|
||
if logErr := s.completeCardCommandAttempt(ctx, attempt, nil, true); logErr != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, logErr, "终结停机 Integration Log 失败")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
lastErr = callErr
|
||
s.logger.Warn("调用停机接口失败，准备重试",
|
||
zap.Int("attempt", i+1),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Error(callErr))
|
||
}
|
||
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"未完成", attemptObserver.auditResult(lastErr), lastIntegrationID,
|
||
map[string]any{"network_status": card.NetworkStatus, "stop_reason": card.StopReason},
|
||
map[string]any{"requested_network_status": constants.NetworkStatusOffline, "stop_reason": stopReason}, lastErr)
|
||
|
||
return lastErr
|
||
}
|
||
|
||
// resumeCardWithRetry 调用运营商复机接口（带重试机制）。
|
||
func (s *StopResumeService) resumeCardWithRetry(ctx context.Context, card *model.IotCard, actionCode, summary string) (*gatewayAttempt, error) {
|
||
if s.gatewayClient == nil {
|
||
failErr := errors.New(errors.CodeInternalError, "Gateway 未配置，停复机操作不可用")
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"失败", constants.AuditResultFailed, "", cardSnapshot(card), nil, failErr)
|
||
return nil, failErr
|
||
}
|
||
|
||
s.logger.Info("调用网关复机",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID))
|
||
|
||
seriesKey := cardCommandSeriesKey(ctx)
|
||
attemptObserver := &cardCommandAttemptObserver{
|
||
service: s, card: card, operation: constants.IntegrationOperationGatewayStartCard,
|
||
scene: constants.CardObservationSceneBusinessResume, seriesKey: seriesKey,
|
||
}
|
||
gatewayCtx := gateway.WithAttemptObserver(ctx, attemptObserver)
|
||
var lastErr error
|
||
lastIntegrationID := ""
|
||
for i := 0; i < s.maxRetries; i++ {
|
||
if i > 0 {
|
||
s.logger.Debug("重试调用复机接口",
|
||
zap.Int("attempt", i+1),
|
||
zap.String("iccid", card.ICCID))
|
||
time.Sleep(s.retryInterval)
|
||
}
|
||
|
||
req := &gateway.CardOperationReq{CardNo: card.ICCID}
|
||
if strings.TrimSpace(card.GatewayExtend) == constants.GatewayCardExtendMachineSeparated {
|
||
req.Extend = constants.GatewayCardStartExtendMachineSeparated
|
||
}
|
||
callErr := s.gatewayClient.StartCard(gatewayCtx, req)
|
||
lastIntegrationID = attemptObserver.lastIntegrationID
|
||
if attemptObserver.recordingErr != nil {
|
||
s.logger.Error("记录复机 Integration Log 失败", zap.String("integration_id", lastIntegrationID), zap.Error(attemptObserver.recordingErr))
|
||
lastErr = attemptObserver.lastCallErr
|
||
if lastErr == nil {
|
||
lastErr = attemptObserver.recordingErr
|
||
}
|
||
break
|
||
}
|
||
if callErr == nil {
|
||
s.logger.Info("网关复机成功",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID))
|
||
return attemptObserver.successful, nil
|
||
}
|
||
|
||
lastErr = callErr
|
||
s.logger.Warn("调用复机接口失败，准备重试",
|
||
zap.Int("attempt", i+1),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Error(callErr))
|
||
}
|
||
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"未完成", attemptObserver.auditResult(lastErr), lastIntegrationID,
|
||
map[string]any{"network_status": card.NetworkStatus, "stop_reason": card.StopReason, "gateway_extend": card.GatewayExtend},
|
||
map[string]any{"requested_network_status": constants.NetworkStatusOnline}, lastErr)
|
||
return nil, lastErr
|
||
}
|
||
|
||
// StartMachineSeparatedCard 对机卡分离停机卡执行复机
|
||
// 调用上游复机前检查本地落库的 Gateway extend，只有“机卡分离停机”才允许复机。
|
||
func (s *StopResumeService) StartMachineSeparatedCard(ctx context.Context, card *model.IotCard) error {
|
||
if card == nil {
|
||
return errors.New(errors.CodeInvalidParam)
|
||
}
|
||
actionCode, summary := constants.AuditActionIotCardOpenAPIStarted, "OpenAPI 恢复 IoT 卡网络"
|
||
if s.gatewayClient == nil {
|
||
failErr := errors.New(errors.CodeInternalError, "Gateway 未配置，停复机操作不可用")
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"失败", constants.AuditResultFailed, "", cardSnapshot(card), nil, failErr)
|
||
return failErr
|
||
}
|
||
|
||
gatewayExtend := strings.TrimSpace(card.GatewayExtend)
|
||
|
||
// 独立卡处于风险停机或已销户状态时，优先返回明确的拒绝原因
|
||
if card.IsStandalone && isRiskGatewayExtend(gatewayExtend) {
|
||
var denyMsg string
|
||
if gatewayExtend == constants.GatewayCardExtendRiskStop {
|
||
denyMsg = "该卡已被运营商风险停机，不允许复机"
|
||
} else {
|
||
denyMsg = "该卡已被运营商销户，不允许复机"
|
||
}
|
||
denyErr := errors.New(errors.CodeForbidden, denyMsg)
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"被拒绝", constants.AuditResultDenied, "", cardSnapshot(card), nil, denyErr)
|
||
return denyErr
|
||
}
|
||
|
||
if gatewayExtend != constants.GatewayCardExtendMachineSeparated {
|
||
denyErr := errors.New(errors.CodeForbidden, constants.AgentOpenAPIResumeOnlyMachineSeparatedMessage)
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"被拒绝", constants.AuditResultDenied, "",
|
||
cardSnapshot(card), map[string]any{"gateway_extend": gatewayExtend}, denyErr)
|
||
return denyErr
|
||
}
|
||
|
||
attempt, err := s.resumeCardWithRetry(ctx, card, actionCode, summary)
|
||
if err != nil {
|
||
wrapErr := errors.Wrap(errors.CodeGatewayError, err, "调用运营商复机失败，请稍后重试")
|
||
return wrapErr
|
||
}
|
||
|
||
now := time.Now()
|
||
if err := s.updateCardAndAppendNetworkSeries(ctx, card, map[string]any{
|
||
"network_status": constants.NetworkStatusOnline,
|
||
"resumed_at": now,
|
||
"stop_reason": "",
|
||
"gateway_extend": "",
|
||
}, constants.CardObservationSceneBusinessResume, "online", uuid.NewString(), actionCode, summary, attempt.log.IntegrationID); err != nil {
|
||
if logErr := s.completeCardCommandAttempt(ctx, attempt, nil, false); logErr != nil {
|
||
s.logger.Error("终结复机 Integration Log 失败", zap.String("integration_id", attempt.log.IntegrationID), zap.Error(logErr))
|
||
}
|
||
wrapErr := errors.Wrap(errors.CodeDatabaseError, err, "更新卡状态失败")
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"结果待核对", constants.AuditResultUnknown,
|
||
attempt.log.IntegrationID, cardSnapshot(card), map[string]any{"requested_network_status": constants.NetworkStatusOnline}, wrapErr)
|
||
return wrapErr
|
||
}
|
||
|
||
s.reschedulePolling(ctx, card.ID)
|
||
if logErr := s.completeCardCommandAttempt(ctx, attempt, nil, true); logErr != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, logErr, "终结 OpenAPI 复机 Integration Log 失败")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ManualStopCard 手动停机单张卡（通过ICCID）
|
||
func (s *StopResumeService) ManualStopCard(ctx context.Context, iccid string) error {
|
||
card, err := s.iotCardStore.GetByICCID(ctx, iccid)
|
||
if err != nil {
|
||
return errors.New(errors.CodeNotFound, "卡不存在")
|
||
}
|
||
actionCode, summary := stopAuditAction(ctx, constants.StopReasonManual)
|
||
|
||
if !s.isRealnameOK(card) {
|
||
denyErr := errors.New(errors.CodeForbidden, "卡未实名，无法操作")
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"被拒绝", constants.AuditResultDenied, "", cardSnapshot(card), nil, denyErr)
|
||
return denyErr
|
||
}
|
||
|
||
// 检查绑定设备是否在复机保护期
|
||
if s.deviceSimBindingStore != nil && s.redis != nil {
|
||
binding, bindErr := s.deviceSimBindingStore.GetActiveBindingByCardID(ctx, card.ID)
|
||
if bindErr == nil && binding != nil {
|
||
exists, _ := s.redis.Exists(ctx, constants.RedisDeviceProtectKey(binding.DeviceID, "start")).Result()
|
||
if exists > 0 {
|
||
denyErr := errors.New(errors.CodeForbidden, "设备复机保护期内，禁止停机")
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"被拒绝", constants.AuditResultDenied, "",
|
||
cardSnapshot(card), map[string]any{"device_id": binding.DeviceID}, denyErr)
|
||
return denyErr
|
||
}
|
||
} else if bindErr != nil && !stderrors.Is(bindErr, gorm.ErrRecordNotFound) {
|
||
wrapErr := errors.Wrap(errors.CodeInternalError, bindErr, "查询卡绑定关系失败")
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"失败", constants.AuditResultFailed, "", cardSnapshot(card), nil, wrapErr)
|
||
return wrapErr
|
||
}
|
||
}
|
||
|
||
if err := s.stopCardWithRetry(ctx, card, constants.StopReasonManual); err != nil {
|
||
return err
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// ManualStartCard 手动复机单张卡（通过ICCID）
|
||
func (s *StopResumeService) ManualStartCard(ctx context.Context, iccid string) error {
|
||
card, err := s.iotCardStore.GetByICCID(ctx, iccid)
|
||
if err != nil {
|
||
return errors.New(errors.CodeNotFound, "卡不存在")
|
||
}
|
||
actionCode, summary := constants.AuditActionIotCardManualStarted, "人工恢复 IoT 卡网络"
|
||
|
||
// 独立卡处于风险停机或已销户状态时，拒绝复机
|
||
if card.IsStandalone && isRiskGatewayExtend(card.GatewayExtend) {
|
||
var denyMsg string
|
||
if strings.TrimSpace(card.GatewayExtend) == constants.GatewayCardExtendRiskStop {
|
||
denyMsg = "该卡已被运营商风险停机，不允许复机"
|
||
} else {
|
||
denyMsg = "该卡已被运营商销户，不允许复机"
|
||
}
|
||
denyErr := errors.New(errors.CodeForbidden, denyMsg)
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"被拒绝", constants.AuditResultDenied, "", cardSnapshot(card), nil, denyErr)
|
||
return denyErr
|
||
}
|
||
|
||
if !s.isRealnameOK(card) {
|
||
denyErr := errors.New(errors.CodeForbidden, "卡未实名，无法操作")
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"被拒绝", constants.AuditResultDenied, "", cardSnapshot(card), nil, denyErr)
|
||
return denyErr
|
||
}
|
||
|
||
// 检查绑定设备是否在停机保护期
|
||
if s.deviceSimBindingStore != nil && s.redis != nil {
|
||
binding, bindErr := s.deviceSimBindingStore.GetActiveBindingByCardID(ctx, card.ID)
|
||
if bindErr == nil && binding != nil {
|
||
exists, _ := s.redis.Exists(ctx, constants.RedisDeviceProtectKey(binding.DeviceID, "stop")).Result()
|
||
if exists > 0 {
|
||
denyErr := errors.New(errors.CodeForbidden, "设备停机保护期内，禁止复机")
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"被拒绝", constants.AuditResultDenied, "",
|
||
cardSnapshot(card), map[string]any{"device_id": binding.DeviceID}, denyErr)
|
||
return denyErr
|
||
}
|
||
} else if bindErr != nil && !stderrors.Is(bindErr, gorm.ErrRecordNotFound) {
|
||
wrapErr := errors.Wrap(errors.CodeInternalError, bindErr, "查询卡绑定关系失败")
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"失败", constants.AuditResultFailed, "", cardSnapshot(card), nil, wrapErr)
|
||
return wrapErr
|
||
}
|
||
}
|
||
|
||
attempt, err := s.resumeCardWithRetry(ctx, card, actionCode, summary)
|
||
if err != nil {
|
||
wrapErr := errors.Wrap(errors.CodeGatewayError, err, "调用运营商复机失败，请稍后重试")
|
||
return wrapErr
|
||
}
|
||
|
||
now := time.Now()
|
||
if err := s.updateCardAndAppendNetworkSeries(ctx, card, map[string]any{
|
||
"network_status": constants.NetworkStatusOnline,
|
||
"resumed_at": now,
|
||
"stop_reason": "",
|
||
}, constants.CardObservationSceneBusinessResume, "online", uuid.NewString(), actionCode, summary, attempt.log.IntegrationID); err != nil {
|
||
if logErr := s.completeCardCommandAttempt(ctx, attempt, nil, false); logErr != nil {
|
||
s.logger.Error("终结复机 Integration Log 失败", zap.String("integration_id", attempt.log.IntegrationID), zap.Error(logErr))
|
||
}
|
||
wrapErr := errors.Wrap(errors.CodeDatabaseError, err, "更新卡状态失败")
|
||
s.recordCardCommandAudit(ctx, card, actionCode, summary+"结果待核对", constants.AuditResultUnknown,
|
||
attempt.log.IntegrationID, cardSnapshot(card), map[string]any{"requested_network_status": constants.NetworkStatusOnline}, wrapErr)
|
||
return wrapErr
|
||
}
|
||
|
||
s.reschedulePolling(ctx, card.ID)
|
||
if logErr := s.completeCardCommandAttempt(ctx, attempt, nil, true); logErr != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, logErr, "终结人工复机 Integration Log 失败")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *StopResumeService) updateCardAndAppendNetworkSeries(
|
||
ctx context.Context,
|
||
card *model.IotCard,
|
||
fields map[string]any,
|
||
scene, expected, operationID, actionCode, summary, integrationID string,
|
||
) error {
|
||
if s.db == nil || s.observationSeriesEvents == nil || card == nil || card.ID == 0 {
|
||
return errors.New(errors.CodeInternalError, "停复机观测 Outbox 能力未配置")
|
||
}
|
||
requestID := requestIDFromContext(ctx)
|
||
if requestID == "" {
|
||
requestID = operationID
|
||
}
|
||
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
if err := tx.Model(&model.IotCard{}).Where("id = ?", card.ID).Updates(fields).Error; err != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "更新卡停复机状态失败")
|
||
}
|
||
if err := s.appendCardCommandAudit(ctx, tx, card, actionCode, summary, constants.AuditResultSuccess,
|
||
integrationID,
|
||
map[string]any{"network_status": card.NetworkStatus, "stop_reason": card.StopReason, "gateway_extend": card.GatewayExtend},
|
||
fields, nil); err != nil {
|
||
return err
|
||
}
|
||
if cardObservationApp.IsSeriesTriggerSuppressed(ctx) {
|
||
return nil
|
||
}
|
||
return s.observationSeriesEvents.AppendSeriesRequested(ctx, tx, cardObservationApp.SeriesRequestedEvent{
|
||
EventID: "card-observation:network-command:" + operationID,
|
||
Scene: scene, ResourceType: constants.CardObservationResourceTypeCard, ResourceID: card.ID,
|
||
SyncTypes: []string{constants.CardObservationSyncTypeNetwork}, ExpectedValue: expected,
|
||
Source: constants.CardObservationSourceBusinessEvent, OccurredAt: time.Now().UTC(),
|
||
RequestID: requestID, CorrelationID: requestID,
|
||
})
|
||
})
|
||
}
|
||
|
||
// invalidatePollingCache 删除轮询卡缓存，下次轮询自动从 DB 重建
|
||
func (s *StopResumeService) invalidatePollingCache(ctx context.Context, cardID uint) {
|
||
if s.redis == nil {
|
||
return
|
||
}
|
||
key := constants.RedisPollingCardInfoKey(cardID)
|
||
if err := s.redis.Del(ctx, key).Err(); err != nil {
|
||
s.logger.Warn("删除轮询卡缓存失败", zap.Uint("card_id", cardID))
|
||
}
|
||
}
|
||
|
||
// reschedulePolling 按卡最新状态重排轮询队列。
|
||
// 未注入生命周期回调时退化为只清缓存，保持原有行为。
|
||
func (s *StopResumeService) reschedulePolling(ctx context.Context, cardID uint) {
|
||
if s.pollingCallback == nil {
|
||
s.invalidatePollingCache(ctx, cardID)
|
||
return
|
||
}
|
||
s.pollingCallback.OnCardStatusChanged(ctx, cardID)
|
||
}
|