junhong_cmp_fiber/internal/application/carrierthreshold/evaluate.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
	}
	// 判定时间取本次观测时刻；观测时刻缺失时退化为系统时刻，绝不写出零值判定时间。
	judgedAt := evaluation.ObservedAt
	if judgedAt.IsZero() {
		judgedAt = s.now()
	}
	// 命中数值冻结为字面量：读数与阈值共用同一对副本，避免把可变的配置读回留痕。
	hitTrafficMB := evaluation.ReadingMB
	hitThresholdValue := threshold.Value
	lock := &model.CarrierTrafficThresholdLock{
		CarrierID:    evaluation.CarrierID,
		CardID:       evaluation.CardID,
		PeriodStart:  periodStart,
		Status:       domain.LockStatusLocked,
		StopStatus:   domain.TaskStatusPending,
		ResumeStatus: domain.TaskStatusPending,
		// 命中事实与停机锁是同一条 INSERT 的列，不会因业务数据导致留痕失败而阻断停机：
		// 命中列的取值范围由判定前置条件（读数已接受、阈值有效、已判定达量）保证，业务数据无法使
		// hit_* CHECK 失败。数据库级故障与既有流量观测一致地整体回滚，停机顺延到下一次观测重试，
		// 不会出现「已提交停机但无留痕」的半事实。
		// 读数取本次判定使用的网关累计读数（与停机判定同一来源，不是本地用量或套餐真流量）；
		// 阈值与单位取本次判定实际生效的通道配置值，冻结命中时值，之后改阈值不改写历史命中。
		HitTrafficMB:      &hitTrafficMB,
		HitThresholdValue: &hitThresholdValue,
		HitThresholdUnit:  threshold.Unit,
		JudgedAt:          judgedAt,
		// 来源按列约束归一：未归类来源记为未记录，避免留痕列 CHECK 反过来阻断停机。
		TriggerSource: domain.SafeTriggerSource(evaluation.Source),
	}
	// 唯一冲突（该周期已处理）在 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.Float64("threshold_value", hitThresholdValue), zap.String("threshold_unit", threshold.Unit),
		zap.Time("judged_at", judgedAt), zap.String("trigger_source", lock.TriggerSource),
		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)
}