iotlcm/admin-svr/server/internal/logic/trafficusage/traffic_usage_pool.go

package trafficusage

import (
	"context"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/dao"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/model"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/service"
	"time"

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

type sTrafficPoolUsageSync struct {
}

type trafficPoolCardRow struct {
	Iccid        string `gconv:"iccid"`
	FlowPoolCode string `gconv:"flow_pool_code"`
}

func NewTrafficPoolUsageSync() *sTrafficPoolUsageSync {
	return &sTrafficPoolUsageSync{}
}

func init() {
	service.RegisterTrafficPoolUsageSync(NewTrafficPoolUsageSync())
}

// TODO: 流量池查询逻辑还未确定
// SyncDaily 同步流量池的流量使用记录
func (s *sTrafficPoolUsageSync) SyncDaily(ctx context.Context) {
	yesterday := time.Now().AddDate(0, 0, -1)
	date := time.Date(yesterday.Year(), yesterday.Month(), yesterday.Day(), 0, 0, 0, 0, time.Local)

	cards, err := queryTrafficPoolCards(ctx)
	if err != nil {
		g.Log().Errorf(ctx, "SyncTrafficPoolUsageDaily failed, date:%s, err:%+v", date.Format("2006-01-02"), err)
		return
	}
	if len(cards) == 0 {
		g.Log().Debug(ctx, "SyncTrafficPoolUsageDaily skipped, no flow pool cards")
		return
	}

	aggregates, err := service.TrafficUsage().AggregatePoolUsageDaily(ctx, date, cards)
	if err != nil {
		g.Log().Errorf(ctx, "SyncTrafficPoolUsageDaily failed, date:%s, err:%+v", date.Format("2006-01-02"), err)
		return
	}
	if items := buildTrafficPoolUsageItems(aggregates, date); len(items) > 0 {
		if err = service.TrafficUsage().WritePoolUsageDaily(ctx, items); err != nil {
			g.Log().Errorf(ctx, "SyncTrafficPoolUsageDaily failed, date:%s, err:%+v", date.Format("2006-01-02"), err)
			return
		}
	}
	if countItems := buildTrafficPoolCardCountItems(cards, date); len(countItems) > 0 {
		if err = service.TrafficUsage().WritePoolCardCountDaily(ctx, countItems); err != nil {
			g.Log().Errorf(ctx, "SyncTrafficPoolUsageDaily failed, date:%s, err:%+v", date.Format("2006-01-02"), err)
			return
		}
	}
}

// SyncMonthly 同步流量池的流量使用记录
func (s *sTrafficPoolUsageSync) SyncMonthly(ctx context.Context) {
	lastMonth := time.Now().AddDate(0, -1, 0)
	month := time.Date(lastMonth.Year(), lastMonth.Month(), 1, 0, 0, 0, 0, time.Local)

	cards, err := queryTrafficPoolCards(ctx)
	if err != nil {
		g.Log().Errorf(ctx, "SyncTrafficPoolUsageMonthly failed, month:%s, err:%+v", month.Format("2006-01"), err)
		return
	}
	if len(cards) == 0 {
		g.Log().Debug(ctx, "SyncTrafficPoolUsageMonthly skipped, no flow pool cards")
		return
	}

	aggregates, err := service.TrafficUsage().AggregatePoolUsageMonthly(ctx, month, cards)
	if err != nil {
		g.Log().Errorf(ctx, "SyncTrafficPoolUsageMonthly failed, month:%s, err:%+v", month.Format("2006-01"), err)
		return
	}
	if items := buildTrafficPoolUsageItems(aggregates, month); len(items) > 0 {
		if err = service.TrafficUsage().WritePoolUsageMonthly(ctx, items); err != nil {
			g.Log().Errorf(ctx, "SyncTrafficPoolUsageMonthly failed, month:%s, err:%+v", month.Format("2006-01"), err)
			return
		}
	}
	if countItems := buildTrafficPoolCardCountItems(cards, month); len(countItems) > 0 {
		if err = service.TrafficUsage().WritePoolCardCountMonthly(ctx, countItems); err != nil {
			g.Log().Errorf(ctx, "SyncTrafficPoolUsageMonthly failed, month:%s, err:%+v", month.Format("2006-01"), err)
			return
		}
	}
}

func queryTrafficPoolCards(ctx context.Context) ([]model.TrafficPoolCard, error) {
	rows := make([]trafficPoolCardRow, 0)
	cardTable := dao.TSimCardInfo.Table()
	columns := dao.TSimCardInfo.Columns()
	if err := trafficUsageEligibleCardModel(ctx).
		Fields(cardTable+"."+columns.Iccid+" AS iccid", cardTable+"."+columns.FlowPoolCode+" AS flow_pool_code").
		Where(cardTable+"."+columns.IsFlowPool, 1).
		Where(cardTable+"."+columns.FlowPoolCode+" <> ?", "").
		Scan(&rows); err != nil {
		return nil, err
	}

	cards := make([]model.TrafficPoolCard, 0, len(rows))
	for _, row := range rows {
		if row.Iccid == "" || row.FlowPoolCode == "" {
			continue
		}
		cards = append(cards, model.TrafficPoolCard{
			Iccid:        row.Iccid,
			FlowPoolCode: row.FlowPoolCode,
		})
	}
	return cards, nil
}

func buildTrafficPoolUsageItems(aggregates []model.TrafficPoolUsageAggregate, date time.Time) []model.TrafficPoolUsageItem {
	items := make([]model.TrafficPoolUsageItem, 0, len(aggregates))
	for _, aggregate := range aggregates {
		if aggregate.FlowPoolCode == "" {
			continue
		}
		items = append(items, model.TrafficPoolUsageItem{
			FlowPoolCode: aggregate.FlowPoolCode,
			UsageMb:      aggregate.UsageMb,
			Date:         date,
		})
	}
	return items
}

func buildTrafficPoolCardCountItems(cards []model.TrafficPoolCard, date time.Time) []model.TrafficPoolCardCountItem {
	countMap := make(map[string]int64)
	for _, card := range cards {
		if card.FlowPoolCode == "" {
			continue
		}
		countMap[card.FlowPoolCode]++
	}

	items := make([]model.TrafficPoolCardCountItem, 0, len(countMap))
	for flowPoolCode, cardCount := range countMap {
		items = append(items, model.TrafficPoolCardCountItem{
			FlowPoolCode: flowPoolCode,
			CardCount:    cardCount,
			Date:         date,
		})
	}
	return items
}