MiFaZhan
269 lines
10 KiB
Go
269 lines
10 KiB
Go
package trafficusage
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"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/library/rabbitmq"
|
|
"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/model"
|
|
"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/model/entity"
|
|
"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/service"
|
|
"time"
|
|
|
|
"github.com/gogf/gf/v2/frame/g"
|
|
"git.5iot.com/iotlcm-public/iotlcm-admin-svr/utility/uuid32"
|
|
)
|
|
|
|
const (
|
|
trafficUsageDailyPageSize = 100
|
|
)
|
|
|
|
type sTrafficUsageDailySync struct {
|
|
}
|
|
|
|
// These indirections keep the trigger deterministic in unit tests without changing
|
|
// the production publisher, clock, or run-id generator.
|
|
var (
|
|
dailyUsageNow = time.Now
|
|
dailyUsageRunID = uuid32.S
|
|
dailyUsageRabbitEnabled = rabbitmq.Enabled
|
|
dailyUsagePublish = rabbitmq.Publish
|
|
)
|
|
|
|
func NewTrafficUsageDailySync() *sTrafficUsageDailySync {
|
|
return &sTrafficUsageDailySync{}
|
|
}
|
|
|
|
func init() {
|
|
service.RegisterTrafficUsageDailySync(NewTrafficUsageDailySync())
|
|
}
|
|
|
|
// StartDaily publishes one command for Worker to execute the previous-day usage
|
|
// full sync. Admin deliberately does not scan cards or call an upstream service;
|
|
// Worker owns pacing, retries, the failure pool and replay.
|
|
func (s *sTrafficUsageDailySync) StartDaily(ctx context.Context) error {
|
|
_, err := s.StartDailyRun(ctx)
|
|
return err
|
|
}
|
|
|
|
func (s *sTrafficUsageDailySync) StartDailyRun(ctx context.Context) (string, error) {
|
|
return s.StartDailyRunScoped(ctx, nil)
|
|
}
|
|
|
|
// StartDailyRunScoped starts a run restricted to scope.Channels. A nil scope, or
|
|
// one whose channels normalize to empty, starts a full sync.
|
|
func (s *sTrafficUsageDailySync) StartDailyRunScoped(ctx context.Context, scope *model.SyncScope) (string, error) {
|
|
return s.StartDailyRunScopedAt(ctx, scope, "")
|
|
}
|
|
|
|
// StartDailyRunScopedAt starts a run for the given day (yyyyMMdd). An empty
|
|
// queryDate keeps the historical behaviour and covers the previous calendar day
|
|
// in the business timezone; a non-empty value lets an operator backfill one
|
|
// specific past day.
|
|
func (s *sTrafficUsageDailySync) StartDailyRunScopedAt(ctx context.Context, scope *model.SyncScope, queryDate string) (string, error) {
|
|
if ctx == nil {
|
|
ctx = context.Background()
|
|
}
|
|
startedAt := time.Now()
|
|
normalized, err := model.NormalizeScope(scope)
|
|
if err != nil {
|
|
g.Log().Errorf(ctx, "event=daily_usage_trigger status=failed stage=validate durationMs=%d error=%q", elapsedMilliseconds(startedAt), err.Error())
|
|
return "", err
|
|
}
|
|
if !dailyUsageRabbitEnabled(ctx) {
|
|
err := fmt.Errorf("daily usage start skipped: rabbitmq is disabled")
|
|
g.Log().Errorf(ctx, "event=daily_usage_trigger status=failed stage=precheck durationMs=%d reason=%q", elapsedMilliseconds(startedAt), err.Error())
|
|
return "", err
|
|
}
|
|
|
|
location, err := time.LoadLocation("Asia/Shanghai")
|
|
if err != nil {
|
|
g.Log().Errorf(ctx, "event=daily_usage_trigger status=failed stage=timezone durationMs=%d error=%q", elapsedMilliseconds(startedAt), err.Error())
|
|
return "", fmt.Errorf("load daily usage timezone: %w", err)
|
|
}
|
|
resolvedDate, err := resolveDailyQueryDate(dailyUsageNow().In(location), location, queryDate)
|
|
if err != nil {
|
|
g.Log().Errorf(ctx, "event=daily_usage_trigger status=failed stage=validate reason=invalid_query_date queryDate=%q durationMs=%d error=%q", queryDate, elapsedMilliseconds(startedAt), err.Error())
|
|
return "", err
|
|
}
|
|
message := model.TrafficUsageDailyStartMessage{
|
|
RunID: fmt.Sprintf("daily-usage-%s-%s", resolvedDate, dailyUsageRunID()),
|
|
Operation: model.OperationPreviousDayUsageFullSync,
|
|
QueryDate: resolvedDate,
|
|
TriggeredAt: dailyUsageNow().In(location).Format(time.RFC3339),
|
|
Scope: normalized,
|
|
}
|
|
if err = dailyUsagePublish(ctx, consts.QueueTrafficUsageDailyStart, message); err != nil {
|
|
g.Log().Errorf(ctx, "event=daily_usage_trigger status=failed stage=publish runId=%s queryDate=%s durationMs=%d error=%q", message.RunID, message.QueryDate, elapsedMilliseconds(startedAt), err.Error())
|
|
return "", fmt.Errorf("publish daily usage start: %w", err)
|
|
}
|
|
g.Log().Infof(ctx, "event=daily_usage_trigger status=published runId=%s queryDate=%s triggeredAt=%s durationMs=%d", message.RunID, message.QueryDate, message.TriggeredAt, elapsedMilliseconds(startedAt))
|
|
return message.RunID, nil
|
|
}
|
|
|
|
// minBackfillDay is the earliest day a manual backfill may target. It only
|
|
// catches obvious typos; it is not a business boundary. Comparing the
|
|
// zero-padded yyyyMMdd strings sorts chronologically.
|
|
const minBackfillDay = "20200101"
|
|
|
|
// resolveDailyQueryDate returns the day a run must cover. An empty requested
|
|
// value keeps the legacy default (the previous day in the business timezone);
|
|
// otherwise the value is validated and returned in canonical yyyyMMdd form.
|
|
func resolveDailyQueryDate(now time.Time, location *time.Location, requested string) (string, error) {
|
|
// The default covers the previous calendar day, which is the same period the
|
|
// legacy admin scan used.
|
|
yesterday := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, location).AddDate(0, 0, -1)
|
|
if requested == "" {
|
|
return yesterday.Format("20060102"), nil
|
|
}
|
|
parsed, err := time.ParseInLocation("20060102", requested, location)
|
|
if err != nil {
|
|
return "", fmt.Errorf("invalid queryDate %q: want yyyyMMdd", requested)
|
|
}
|
|
if parsed.After(yesterday) {
|
|
return "", fmt.Errorf("invalid queryDate %q: must not be later than %s", requested, yesterday.Format("20060102"))
|
|
}
|
|
if parsed.Format("20060102") < minBackfillDay {
|
|
return "", fmt.Errorf("invalid queryDate %q: must not be earlier than %s", requested, minBackfillDay)
|
|
}
|
|
return parsed.Format("20060102"), nil
|
|
}
|
|
|
|
// SyncDaily is the legacy in-process scan. It is retained as the rollback path
|
|
// until the worker chain is verified in production, and it is no longer
|
|
// registered on any timer.
|
|
func (s *sTrafficUsageDailySync) SyncDaily(ctx context.Context) {
|
|
now := time.Now().In(historyUsageLocation)
|
|
yesterday := now.AddDate(0, 0, -1)
|
|
queryDate := yesterday.Format("20060102")
|
|
date := time.Date(yesterday.Year(), yesterday.Month(), yesterday.Day(), 0, 0, 0, 0, historyUsageLocation)
|
|
|
|
cardColumns := dao.TSimCardInfo.Columns()
|
|
cardTable := dao.TSimCardInfo.Table()
|
|
snapshotMaxID, err := trafficUsageEligibleCardModel(ctx).Max(cardTable + "." + cardColumns.Id)
|
|
if err != nil {
|
|
g.Log().Errorf(ctx, "SyncTrafficUsageDaily snapshot sim cards failed: %+v", err)
|
|
return
|
|
}
|
|
maxID := int64(snapshotMaxID)
|
|
if maxID <= 0 {
|
|
return
|
|
}
|
|
|
|
var lastID int64
|
|
for lastID < maxID {
|
|
list := []entity.TSimCardInfo{}
|
|
if err = trafficUsageEligibleCardModel(ctx).
|
|
Fields(trafficUsageCardFields()).
|
|
WhereGT(cardTable+"."+cardColumns.Id, lastID).
|
|
WhereLTE(cardTable+"."+cardColumns.Id, maxID).
|
|
OrderAsc(cardTable + "." + cardColumns.Id).
|
|
Limit(trafficUsageDailyPageSize).
|
|
Scan(&list); err != nil {
|
|
g.Log().Errorf(ctx, "SyncTrafficUsageDaily list sim cards failed, lastId:%d, err:%+v", lastID, err)
|
|
return
|
|
}
|
|
if len(list) == 0 {
|
|
return
|
|
}
|
|
lastID = list[len(list)-1].Id
|
|
|
|
cardMap := trafficUsageDailyCardMap(list)
|
|
simKeys := trafficUsageDailySimKeys(list)
|
|
output, err := service.UpstreamChan().QuerySimDatausageDaily(ctx, &model.SimDataUsageDailyInput{
|
|
SimKeys: simKeys,
|
|
ReqData: model.SimDataUsageDailyInputItem{
|
|
QueryDate: queryDate,
|
|
},
|
|
})
|
|
if err != nil {
|
|
g.Log().Errorf(ctx, "SyncTrafficUsageDaily query daily failed, lastId:%d, count:%d, err:%+v", lastID, len(simKeys), err)
|
|
for _, simKey := range simKeys {
|
|
s.publishRetry(ctx, simKey, cardMap[simKey.ICCID], queryDate)
|
|
}
|
|
continue
|
|
}
|
|
|
|
success, failed := reconcileUsageWithPeriod(simKeys, output, func(item model.SimDataUsageDailyOutputItem) float64 { return item.DataAmount }, func(item model.SimDataUsageDailyOutputItem) string { return item.QueryDate }, normalizeUsageDay, queryDate)
|
|
for _, key := range failed {
|
|
s.publishRetry(ctx, key, cardMap[key.ICCID], queryDate)
|
|
}
|
|
|
|
if items := s.buildDailyItems(success, cardMap, date); len(items) > 0 {
|
|
if err = service.TrafficUsage().UpdateDaily(ctx, items); err != nil {
|
|
g.Log().Errorf(ctx, "SyncTrafficUsageDaily write tdengine failed, lastId:%d, err:%+v", lastID, err)
|
|
for _, item := range success {
|
|
s.publishRetry(ctx, item.SimKey, cardMap[item.SimKey.ICCID], queryDate, item.ResData.DataAmount)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// buildDailyItems 构建SIM卡的流量使用记录
|
|
func (s *sTrafficUsageDailySync) buildDailyItems(output []model.SimBaseRes[model.SimDataUsageDailyOutputItem], cardMap map[string]model.TrafficUsageSimCard, date time.Time) []model.TrafficUsageDailyItem {
|
|
items := make([]model.TrafficUsageDailyItem, 0, len(output))
|
|
for _, item := range output {
|
|
if item.SimKey == nil {
|
|
continue
|
|
}
|
|
card := cardMap[item.SimKey.ICCID]
|
|
items = append(items, model.TrafficUsageDailyItem{
|
|
TrafficUsageBase: model.TrafficUsageBase{
|
|
Iccid: item.SimKey.ICCID,
|
|
UsageMb: item.ResData.DataAmount,
|
|
PackageCode: card.PackageCode,
|
|
},
|
|
Date: date,
|
|
})
|
|
}
|
|
return items
|
|
}
|
|
|
|
// publishRetry 发布重试消息
|
|
func (s *sTrafficUsageDailySync) publishRetry(ctx context.Context, simKey *model.SimKey, card model.TrafficUsageSimCard, queryDate string, usage ...float64) {
|
|
var snapshot *float64
|
|
if len(usage) > 0 {
|
|
snapshot = &usage[0]
|
|
}
|
|
if !rabbitmq.Enabled(ctx) {
|
|
g.Log().Errorf(ctx, "traffic usage daily retry skipped, rabbitmq disabled, simKey:%+v, queryDate:%s", simKey, queryDate)
|
|
return
|
|
}
|
|
if err := rabbitmq.PublishConfirmed(ctx, consts.QueueTrafficUsageDailyRetry, model.TrafficUsageDailyRetryMessage{
|
|
UsageMb: snapshot,
|
|
SimKey: simKey,
|
|
Card: card,
|
|
QueryDate: queryDate,
|
|
}); err != nil {
|
|
g.Log().Errorf(ctx, "publish traffic usage daily retry failed, simKey:%+v, queryDate:%s, err:%+v", simKey, queryDate, err)
|
|
}
|
|
}
|
|
|
|
// trafficUsageDailySimKeys 提取SIM卡的ChanKey、ICCID、IMSI、MSISDN
|
|
func trafficUsageDailySimKeys(list []entity.TSimCardInfo) []*model.SimKey {
|
|
simKeys := make([]*model.SimKey, 0, len(list))
|
|
for _, item := range list {
|
|
simKeys = append(simKeys, &model.SimKey{
|
|
ChanKey: item.ChanKey,
|
|
ICCID: item.Iccid,
|
|
IMSI: item.Imsi,
|
|
MSISDN: item.Msisdn,
|
|
})
|
|
}
|
|
return simKeys
|
|
}
|
|
|
|
// trafficUsageDailyCardMap 提取SIM卡的ChanKey、ICCID、IMSI、MSISDN、PackageCode、FlowPoolCode、IsFlowPool
|
|
func trafficUsageDailyCardMap(list []entity.TSimCardInfo) map[string]model.TrafficUsageSimCard {
|
|
cardMap := make(map[string]model.TrafficUsageSimCard, len(list))
|
|
for _, item := range list {
|
|
cardMap[item.Iccid] = model.TrafficUsageSimCard{
|
|
Iccid: item.Iccid,
|
|
PackageCode: item.CurrentPackageCode,
|
|
}
|
|
}
|
|
return cardMap
|
|
}
|