break
59b3df868a
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 12m59s
108 lines
4.3 KiB
Go
108 lines
4.3 KiB
Go
package carrierthreshold
|
||
|
||
import (
|
||
"context"
|
||
"strconv"
|
||
|
||
"go.uber.org/zap"
|
||
"gorm.io/gorm"
|
||
|
||
domain "github.com/break/junhong_cmp_fiber/internal/domain/carrierthreshold"
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||
)
|
||
|
||
// EvaluateInTx 在流量观测事务内判定通道阈值，达量时写周期锁并同事务写停机事件。
|
||
//
|
||
// 判定前提由调用方保证：只有「读数被接受」（ReadingAccepted）的观测才进入本方法，
|
||
// 异常下降保护命中的观测不参与判定。重复达量观测由部分唯一索引的 23505 识别为
|
||
// 「该周期已处理」并幂等跳过：不重复建锁、不重复写停机事件，该冲突与流量基线 CAS 的
|
||
// CodeConflict 语义不同，绝不能混流。返回错误表示必须整体回滚（与既有流量事实同事务）。
|
||
func (s *Service) EvaluateInTx(ctx context.Context, tx *gorm.DB, evaluation domain.Evaluation) error {
|
||
if s == nil || s.db == nil || s.repository == nil {
|
||
return errors.New(errors.CodeInternalError, "通道流量阈值判定能力未完整配置")
|
||
}
|
||
if tx == nil {
|
||
return errors.New(errors.CodeInvalidStatus, "通道流量阈值判定必须传入事务句柄")
|
||
}
|
||
if evaluation.CardID == 0 || evaluation.CarrierID == 0 {
|
||
return nil
|
||
}
|
||
var carrier model.Carrier
|
||
// 只取判定所需列：周期起点必须来自该运营商的 data_reset_day，漏取会让周期判定失去依据。
|
||
if err := tx.WithContext(ctx).Select("id", "data_reset_day", "traffic_threshold_enabled", "traffic_threshold_value", "traffic_threshold_unit").
|
||
Where("id = ?", evaluation.CarrierID).First(&carrier).Error; err != nil {
|
||
if err == gorm.ErrRecordNotFound {
|
||
// 卡引用的运营商已不存在：不判定、不建锁，保持与既有卡事实不一致的现状。
|
||
return nil
|
||
}
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "查询运营商通道流量阈值配置失败")
|
||
}
|
||
threshold := thresholdOf(carrier)
|
||
if !threshold.Enabled {
|
||
return nil
|
||
}
|
||
if !threshold.Valid() {
|
||
// 配置半残（启用但无数值/单位，或单位未知）时绝不按 0 判定，跳过并留可观测日志。
|
||
s.logger.Warn("运营商通道流量阈值配置不完整，跳过达量判定",
|
||
zap.Uint("carrier_id", carrier.ID), zap.String("unit", threshold.Unit), zap.Float64("value", threshold.Value))
|
||
return nil
|
||
}
|
||
reached, err := threshold.Reached(evaluation.ReadingMB)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !reached {
|
||
return nil
|
||
}
|
||
periodStart, err := domain.PeriodStart(evaluation.ObservedAt, carrier.DataResetDay)
|
||
if err != nil {
|
||
s.logger.Warn("运营商上游流量重置日非法，跳过达量判定",
|
||
zap.Uint("carrier_id", carrier.ID), zap.Int("data_reset_day", carrier.DataResetDay))
|
||
return nil
|
||
}
|
||
lock := &model.CarrierTrafficThresholdLock{
|
||
CarrierID: evaluation.CarrierID,
|
||
CardID: evaluation.CardID,
|
||
PeriodStart: periodStart,
|
||
Status: domain.LockStatusLocked,
|
||
StopStatus: domain.TaskStatusPending,
|
||
ResumeStatus: domain.TaskStatusPending,
|
||
}
|
||
// 唯一冲突（该周期已处理）在 createLockInTx 内以保存点隔离：不重复建锁、不重复写停机事件。
|
||
created, err := s.createLockInTx(ctx, tx, lock)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !created {
|
||
s.logger.Info("该计费周期已存在通道阈值停机锁，跳过重复判定",
|
||
zap.Uint("carrier_id", evaluation.CarrierID), zap.Uint("card_id", evaluation.CardID))
|
||
return nil
|
||
}
|
||
if err := AppendStopRequested(ctx, tx, s.repository, lock); err != nil {
|
||
return err
|
||
}
|
||
s.logger.Info("卡流量达到运营商通道阈值，已写周期锁与停机事件",
|
||
zap.Uint("carrier_id", evaluation.CarrierID), zap.Uint("card_id", evaluation.CardID),
|
||
zap.Uint("lock_id", lock.ID), zap.Float64("reading_mb", evaluation.ReadingMB),
|
||
zap.Time("period_start", periodStart))
|
||
return nil
|
||
}
|
||
|
||
// thresholdOf 把运营商持久化列转换为领域阈值配置。
|
||
func thresholdOf(carrier model.Carrier) domain.Threshold {
|
||
threshold := domain.Threshold{
|
||
Enabled: carrier.TrafficThresholdEnabled == 1,
|
||
Unit: carrier.TrafficThresholdUnit,
|
||
}
|
||
if carrier.TrafficThresholdValue != nil {
|
||
threshold.Value = *carrier.TrafficThresholdValue
|
||
}
|
||
return threshold
|
||
}
|
||
|
||
// lockKeyValue 返回周期锁在 Outbox 事件中的稳定字符串标识。
|
||
func lockKeyValue(lockID uint) string {
|
||
return strconv.FormatUint(uint64(lockID), 10)
|
||
}
|