iotlcm/admin-svr/server/internal/logic/trafficcutoff/scanner.go

package trafficcutoff

import (
	"context"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/consts"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/dao"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/logic/cardflow"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/service"
	"math"
	"strings"
	"time"

	"github.com/gogf/gf/v2/frame/g"
)

type sTrafficCutoff struct{}

func NewTrafficCutoff() *sTrafficCutoff {
	return &sTrafficCutoff{}
}

func init() {
	service.RegisterTrafficCutoff(NewTrafficCutoff())
}

func (s *sTrafficCutoff) Check(ctx context.Context) {
	if ctx == nil {
		ctx = context.Background()
	}
	runTime := time.Now()
	stats := cutoffStats{}
	lastID := int64(0)

	g.Log().Infof(ctx, "TrafficCutoff.Check begin, runTime:%s", runTime.Format("2006-01-02 15:04:05"))
	for {
		if err := ctx.Err(); err != nil {
			stats.ErrorCount++
			g.Log().Warningf(ctx, "TrafficCutoff.Check canceled, lastID:%d, err:%+v", lastID, err)
			break
		}
		cards, err := s.queryCandidates(ctx, lastID, runTime)
		if err != nil {
			stats.ErrorCount++
			g.Log().Errorf(ctx, "TrafficCutoff.Check query candidates failed, lastID:%d, err:%+v", lastID, err)
			break
		}
		if len(cards) == 0 {
			break
		}
		stats.CandidateCount += len(cards)
		for _, card := range cards {
			if card.Id > lastID {
				lastID = card.Id
			}
		}
		s.handleCandidates(ctx, cards, runTime, &stats)
		if len(cards) < cutoffPageSize {
			break
		}
	}

	g.Log().Infof(
		ctx,
		"TrafficCutoff.Check end, candidates:%d, hits:%d, duplicated:%d, skipped:%d, errors:%d, batches:%d",
		stats.CandidateCount,
		stats.HitCount,
		stats.DupeCount,
		stats.SkipCount,
		stats.ErrorCount,
		stats.BatchCount,
	)
}

func (s *sTrafficCutoff) queryCandidates(ctx context.Context, lastID int64, runTime time.Time) ([]cutoffCard, error) {
	cardTable := dao.TSimCardInfo.Table()
	columns := dao.TSimCardInfo.Columns()
	packageTable := dao.TSimFlowPackageInfo.Table()
	packageColumns := dao.TSimFlowPackageInfo.Columns()
	packageAlias := "current_package"
	crossMonthHoldField := fieldName(cardTable, "traffic_cutoff_cross_month_hold")
	triggeredAtField := fieldName(cardTable, "traffic_cutoff_triggered_at")
	field := func(name string) string {
		return cardTable + "." + name
	}
	cards := make([]cutoffCard, 0, cutoffPageSize)
	err := dao.TSimCardInfo.Ctx(ctx).
		LeftJoin(packageTable+" as "+packageAlias, packageAlias+"."+packageColumns.PackageCode+" = "+field(columns.CurrentPackageCode)+" AND "+packageAlias+"."+packageColumns.TenantId+" = "+field(columns.TenantId)).
		Fields(
			field(columns.Id),
			field(columns.Iccid),
			field(columns.Msisdn),
			field(columns.Imsi),
			field(columns.CurrentPackageCode),
			field(columns.PackageStatus),
			field(columns.BossCardStatus),
			field(columns.IsFlowPool),
			field(columns.PackageTotalFlowMb),
			field(columns.UsedFlowMb),
			field(columns.DataServiceStatus),
			field(columns.TrafficCutoffStatus),
			field(columns.TrafficCutoffThreshold),
			crossMonthHoldField,
			triggeredAtField,
			field(columns.ChanKey),
			field(columns.TenantId),
			field(columns.DeptId),
			field(columns.CreatedDept),
			field(columns.CreatedBy),
			cardflow.TotalFlowExpr(cardTable, packageAlias)+" AS total_flow_mb",
			cardflow.RemainFlowExpr(cardTable, packageAlias)+" AS remain_flow_mb",
		).
		WhereGT(field(columns.Id), lastID).
		Where(field(columns.TrafficCutoffStatus), cutoffStatusEnabled).
		Where("(("+field(columns.TrafficCutoffThreshold)+" > ? AND "+field(columns.UsedFlowMb)+" >= "+field(columns.TrafficCutoffThreshold)+") OR ("+crossMonthHoldField+" = ? AND "+triggeredAtField+" IS NOT NULL AND "+triggeredAtField+" < ?))", 0, 1, monthStart(runTime)).
		Where(field(columns.DataServiceStatus), cutoffDataServiceStatusEnabled).
		Where(field(columns.PackageStatus), consts.PackageStatusEffective).
		Where(field(columns.BossCardStatus), consts.V3ApiSimBossCardStatusActivated).
		Where(field(columns.IsFlowPool), 0).
		Where(field(columns.Iccid)+" <> ?", "").
		Where(field(columns.ChanKey)+" <> ?", "").
		OrderAsc(field(columns.Id)).
		Limit(cutoffPageSize).
		Scan(&cards)
	return cards, err
}

func monthStart(value time.Time) time.Time {
	location := value.Location()
	return time.Date(value.Year(), value.Month(), 1, 0, 0, 0, 0, location)
}

func (s *sTrafficCutoff) handleCandidates(ctx context.Context, cards []cutoffCard, runTime time.Time, stats *cutoffStats) {
	groups := make(map[cutoffBatchGroupKey][]lockedCutoffCard)
	for _, card := range cards {
		if reason := validateCutoffCard(card, runTime); reason != "" {
			stats.SkipCount++
			g.Log().Infof(ctx, "TrafficCutoff.Check skipped, cardId:%d, iccid:%s, reason:%s", card.Id, card.Iccid, reason)
			continue
		}
		acquired, dedupeKey, err := acquireCutoffDedupe(ctx, card, runTime)
		if err != nil {
			stats.ErrorCount++
			g.Log().Errorf(ctx, "TrafficCutoff.Check acquire dedupe failed, cardId:%d, iccid:%s, err:%+v", card.Id, card.Iccid, err)
			continue
		}
		if !acquired {
			stats.DupeCount++
			g.Log().Infof(ctx, "TrafficCutoff.Check duplicated by redis, cardId:%d, iccid:%s", card.Id, card.Iccid)
			continue
		}
		key := cutoffGroupKey(card)
		groups[key] = append(groups[key], lockedCutoffCard{Card: card, DedupeKey: dedupeKey})
	}

	for key, items := range groups {
		if err := ctx.Err(); err != nil {
			stats.ErrorCount++
			releaseCutoffDedupeBatch(ctx, items)
			g.Log().Warningf(ctx, "TrafficCutoff.Check canceled before submit batch, tenantId:%s, cards:%d, err:%+v", key.TenantId, len(items), err)
			continue
		}
		result, err := submitCutoffBatch(ctx, key, items, runTime)
		if err != nil {
			stats.ErrorCount++
			releaseCutoffDedupeBatch(ctx, items)
			g.Log().Errorf(ctx, "TrafficCutoff.Check submit cutoff batch failed, tenantId:%s, dept:%d, user:%d, cards:%d, err:%+v", key.TenantId, key.CreatedDept, key.CreatedBy, len(items), err)
			continue
		}
		stats.HitCount += len(items)
		stats.BatchCount++
		g.Log().Infof(ctx, "TrafficCutoff.Check batch submitted, batchNo:%s, batchType:%d, tenantId:%s, dept:%d, user:%d, cards:%d", result.BatchNo, result.BatchType, key.TenantId, key.CreatedDept, key.CreatedBy, len(items))
		notifyCutoffBatch(ctx, key, items, result.BatchNo, runTime)
	}
}

func validateCutoffCard(card cutoffCard, runTime time.Time) string {
	if card.Id <= 0 {
		return "card id is empty"
	}
	if strings.TrimSpace(card.TenantId) == "" {
		return "tenant id is empty"
	}
	if card.CreatedDept <= 0 || card.CreatedBy <= 0 {
		return "created dept or created by is empty"
	}
	if strings.TrimSpace(card.Iccid) == "" {
		return "iccid is empty"
	}
	if strings.TrimSpace(card.ChanKey) == "" {
		return "chan key is empty"
	}
	if card.TrafficCutoffStatus != cutoffStatusEnabled {
		return "traffic cutoff is not enabled"
	}
	if !card.isThresholdReached() && !card.isCrossMonthHoldReapplyAt(runTime) {
		return "traffic cutoff threshold is not reached"
	}
	if card.DataServiceStatus != cutoffDataServiceStatusEnabled {
		return "data service is not enabled"
	}
	if card.PackageStatus != consts.PackageStatusEffective {
		return "package is not effective"
	}
	if card.BossCardStatus != consts.V3ApiSimBossCardStatusActivated {
		return "boss card status is not activated"
	}
	if card.IsFlowPool != 0 {
		return "flow pool card is not supported"
	}
	return ""
}

func fieldName(table, column string) string {
	return table + "." + column
}

func cutoffGroupKey(card cutoffCard) cutoffBatchGroupKey {
	return cutoffBatchGroupKey{
		TenantId:    strings.TrimSpace(card.TenantId),
		CreatedDept: card.CreatedDept,
		CreatedBy:   card.CreatedBy,
	}
}

func cutoffTotalFlowMb(card cutoffCard) float64 {
	if card.TotalFlowMb > 0 {
		return card.TotalFlowMb
	}
	return card.PackageTotalFlowMb
}

func cutoffRemainFlowMb(card cutoffCard) float64 {
	remain := card.RemainFlowMb
	if remain == 0 && cutoffTotalFlowMb(card) > 0 {
		remain = cutoffTotalFlowMb(card) - card.UsedFlowMb
	}
	if remain < 0 {
		return 0
	}
	return remain
}

func roundCutoffFlow(value float64) float64 {
	return math.Round(value*1000) / 1000
}