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" "git.5iot.com/iotlcm-public/iotlcm-admin-svr/utility/uuid32" "github.com/gogf/gf/v2/frame/g" ) const ( trafficUsageMonthlyPageSize = 100 ) type sTrafficUsageMonthlySync struct { } const monthlyUsageFullSyncOperation = "current_month_usage_full_sync" // These indirections keep the trigger deterministic in unit tests without changing // the production publisher, clock, or run-id generator. var ( monthlyUsageNow = time.Now monthlyUsageRunID = uuid32.S monthlyUsageRabbitEnabled = rabbitmq.Enabled monthlyUsagePublish = rabbitmq.Publish monthlyHistoryNow = time.Now monthlyHistoryRunID = uuid32.S monthlyHistoryRabbitEnabled = rabbitmq.Enabled monthlyHistoryPublish = rabbitmq.Publish ) func NewTrafficUsageMonthlySync() *sTrafficUsageMonthlySync { return &sTrafficUsageMonthlySync{} } func init() { service.RegisterTrafficUsageMonthlySync(NewTrafficUsageMonthlySync()) } // StartMonthly publishes one command for Worker to execute the full monthly // usage sync. Admin deliberately does not scan cards or call an upstream service. func (s *sTrafficUsageMonthlySync) StartMonthly(ctx context.Context) error { _, err := s.StartMonthlyRun(ctx) return err } func (s *sTrafficUsageMonthlySync) StartMonthlyRun(ctx context.Context) (string, error) { return s.StartMonthlyRunScoped(ctx, nil) } // StartMonthlyRunScoped starts a run restricted to scope.Channels. A nil scope, // or one whose channels normalize to empty, starts a full sync. func (s *sTrafficUsageMonthlySync) StartMonthlyRunScoped(ctx context.Context, scope *model.SyncScope) (string, error) { if ctx == nil { ctx = context.Background() } startedAt := time.Now() normalized, err := model.NormalizeScope(scope) if err != nil { g.Log().Errorf(ctx, "event=monthly_usage_trigger status=failed stage=validate durationMs=%d error=%q", elapsedMilliseconds(startedAt), err.Error()) return "", err } if !monthlyUsageRabbitEnabled(ctx) { err := fmt.Errorf("monthly usage start skipped: rabbitmq is disabled") g.Log().Errorf(ctx, "event=monthly_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=monthly_usage_trigger status=failed stage=timezone durationMs=%d error=%q", elapsedMilliseconds(startedAt), err.Error()) return "", fmt.Errorf("load monthly usage timezone: %w", err) } now := monthlyUsageNow().In(location) message := model.TrafficUsageMonthlyStartMessage{ RunID: fmt.Sprintf("current-month-usage-%s-%s", now.Format("20060102"), monthlyUsageRunID()), Operation: monthlyUsageFullSyncOperation, QueryMonth: now.Format("200601"), TriggeredAt: now.Format(time.RFC3339), Scope: normalized, } if err = monthlyUsagePublish(ctx, consts.QueueTrafficUsageMonthlyStart, message); err != nil { g.Log().Errorf(ctx, "event=monthly_usage_trigger status=failed stage=publish runId=%s queryMonth=%s durationMs=%d error=%q", message.RunID, message.QueryMonth, elapsedMilliseconds(startedAt), err.Error()) return "", fmt.Errorf("publish monthly usage start: %w", err) } g.Log().Infof(ctx, "event=monthly_usage_trigger status=published runId=%s queryMonth=%s triggeredAt=%s durationMs=%d", message.RunID, message.QueryMonth, message.TriggeredAt, elapsedMilliseconds(startedAt)) return message.RunID, nil } func elapsedMilliseconds(startedAt time.Time) int64 { return time.Since(startedAt).Milliseconds() } // StartMonthlyHistory publishes one command for Worker to execute the // previous-month usage history sync (the TDengine monthly table). func (s *sTrafficUsageMonthlySync) StartMonthlyHistory(ctx context.Context) error { _, err := s.StartMonthlyHistoryRun(ctx) return err } func (s *sTrafficUsageMonthlySync) StartMonthlyHistoryRun(ctx context.Context) (string, error) { return s.StartMonthlyHistoryRunScoped(ctx, nil) } // StartMonthlyHistoryRunScoped starts a monthly history run restricted to // scope.Channels. A nil scope, or one whose channels normalize to empty, starts // a full sync. // // The period is the previous month, matching what the legacy SyncMonthly scan // wrote. Do not confuse this with StartMonthlyRunScoped, which fills the // current-month snapshot in MySQL. func (s *sTrafficUsageMonthlySync) StartMonthlyHistoryRunScoped(ctx context.Context, scope *model.SyncScope) (string, error) { return s.StartMonthlyHistoryRunScopedAt(ctx, scope, "") } // StartMonthlyHistoryRunScopedAt starts a monthly history run for the given // month (yyyyMM). An empty queryMonth keeps the historical behaviour and covers // the previous calendar month; a non-empty value lets an operator backfill one // specific past month. func (s *sTrafficUsageMonthlySync) StartMonthlyHistoryRunScopedAt(ctx context.Context, scope *model.SyncScope, queryMonth 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=monthly_history_trigger status=failed stage=validate durationMs=%d error=%q", elapsedMilliseconds(startedAt), err.Error()) return "", err } if !monthlyHistoryRabbitEnabled(ctx) { err := fmt.Errorf("monthly history start skipped: rabbitmq is disabled") g.Log().Errorf(ctx, "event=monthly_history_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=monthly_history_trigger status=failed stage=timezone durationMs=%d error=%q", elapsedMilliseconds(startedAt), err.Error()) return "", fmt.Errorf("load monthly history timezone: %w", err) } now := monthlyHistoryNow().In(location) resolvedMonth, err := resolveMonthlyHistoryQueryMonth(now, location, queryMonth) if err != nil { g.Log().Errorf(ctx, "event=monthly_history_trigger status=failed stage=validate reason=invalid_query_month queryMonth=%q durationMs=%d error=%q", queryMonth, elapsedMilliseconds(startedAt), err.Error()) return "", err } message := model.TrafficUsageMonthlyHistoryStartMessage{ RunID: fmt.Sprintf("monthly-history-%s-%s", resolvedMonth, monthlyHistoryRunID()), Operation: model.OperationPreviousMonthUsageFullSync, QueryMonth: resolvedMonth, TriggeredAt: now.Format(time.RFC3339), Scope: normalized, } if err = monthlyHistoryPublish(ctx, consts.QueueTrafficUsageMonthlyHistoryStart, message); err != nil { g.Log().Errorf(ctx, "event=monthly_history_trigger status=failed stage=publish runId=%s queryMonth=%s durationMs=%d error=%q", message.RunID, message.QueryMonth, elapsedMilliseconds(startedAt), err.Error()) return "", fmt.Errorf("publish monthly history start: %w", err) } g.Log().Infof(ctx, "event=monthly_history_trigger status=published runId=%s queryMonth=%s triggeredAt=%s durationMs=%d", message.RunID, message.QueryMonth, message.TriggeredAt, elapsedMilliseconds(startedAt)) return message.RunID, nil } // minBackfillMonth is the earliest month a manual backfill may target. It only // catches obvious typos; it is not a business boundary. const minBackfillMonth = "202001" // resolveMonthlyHistoryQueryMonth returns the month a history run must cover. An // empty requested value keeps the legacy default (the previous calendar month); // otherwise the value is validated and returned in canonical yyyyMM form. func resolveMonthlyHistoryQueryMonth(now time.Time, location *time.Location, requested string) (string, error) { currentMonth := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, location) if requested == "" { // The previous month is what the legacy scan backfilled, and the // month-first day is what the worker's snapshotMonth derives the // end-of-month ts from. return currentMonth.AddDate(0, -1, 0).Format("200601"), nil } parsed, err := time.ParseInLocation("200601", requested, location) if err != nil { return "", fmt.Errorf("invalid queryMonth %q: want yyyyMM", requested) } // The current month is served by the MySQL snapshot chain // (StartMonthlyRunScoped), not by the TDengine history chain. if !parsed.Before(currentMonth) { return "", fmt.Errorf("invalid queryMonth %q: must be earlier than the current month %s", requested, currentMonth.Format("200601")) } if parsed.Format("200601") < minBackfillMonth { return "", fmt.Errorf("invalid queryMonth %q: must not be earlier than %s", requested, minBackfillMonth) } return parsed.Format("200601"), nil } // SyncMonthly is the legacy in-process scan. It is retained as the rollback path // until the worker monthly history chain is verified in production. func (s *sTrafficUsageMonthlySync) SyncMonthly(ctx context.Context) { month := previousUsageMonth(time.Now()) queryDate := month.Format("200601") cardColumns := dao.TSimCardInfo.Columns() cardTable := dao.TSimCardInfo.Table() snapshotMaxID, err := trafficUsageEligibleCardModel(ctx).Max(cardTable + "." + cardColumns.Id) if err != nil { g.Log().Errorf(ctx, "SyncTrafficUsageMonthly 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(trafficUsageMonthlyPageSize). Scan(&list); err != nil { g.Log().Errorf(ctx, "SyncTrafficUsageMonthly list sim cards failed, lastId:%d, err:%+v", lastID, err) return } if len(list) == 0 { return } lastID = list[len(list)-1].Id cardMap := trafficUsageMonthlyCardMap(list) simKeys := trafficUsageMonthlySimKeys(list) output, err := service.UpstreamChan().QuerySimDatausageMonthly(ctx, &model.SimDataUsageMonthlyInput{ SimKeys: simKeys, ReqData: model.SimDataUsageMonthlyInputItem{ QueryDate: queryDate, }, }) if err != nil { g.Log().Errorf(ctx, "SyncTrafficUsageMonthly query monthly 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.SimDataUsageMonthlyOutputItem) float64 { return item.DataAmount }, func(item model.SimDataUsageMonthlyOutputItem) string { return item.DeadLine }, normalizeUsageMonth, queryDate) for _, key := range failed { s.publishRetry(ctx, key, cardMap[key.ICCID], queryDate) } if items := s.buildMonthlyItems(success, cardMap, month); len(items) > 0 { if err = service.TrafficUsage().UpdateMonthly(ctx, items); err != nil { g.Log().Errorf(ctx, "SyncTrafficUsageMonthly 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) } } } } } func (s *sTrafficUsageMonthlySync) buildMonthlyItems(output []model.SimBaseRes[model.SimDataUsageMonthlyOutputItem], cardMap map[string]model.TrafficUsageSimCard, month time.Time) []model.TrafficUsageMonthlyItem { items := make([]model.TrafficUsageMonthlyItem, 0, len(output)) for _, item := range output { if item.SimKey == nil { continue } card := cardMap[item.SimKey.ICCID] items = append(items, model.TrafficUsageMonthlyItem{ TrafficUsageBase: model.TrafficUsageBase{ Iccid: item.SimKey.ICCID, UsageMb: item.ResData.DataAmount, PackageCode: card.PackageCode, }, Month: month, }) } return items } func (s *sTrafficUsageMonthlySync) 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 monthly retry skipped, rabbitmq disabled, simKey:%+v, queryDate:%s", simKey, queryDate) return } if err := rabbitmq.PublishConfirmed(ctx, consts.QueueTrafficUsageMonthlyRetry, model.TrafficUsageMonthlyRetryMessage{ UsageMb: snapshot, SimKey: simKey, Card: card, QueryDate: queryDate, }); err != nil { g.Log().Errorf(ctx, "publish traffic usage monthly retry failed, simKey:%+v, queryDate:%s, err:%+v", simKey, queryDate, err) } } func trafficUsageMonthlySimKeys(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 } func trafficUsageMonthlyCardMap(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 }