package mongo import ( "context" "fmt" "fwh-toolbox-server/database" "fwh-toolbox-server/domain" "time" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/mongo" "go.mongodb.org/mongo-driver/mongo/options" ) const ( collectionSimCardRun = "SimCardDailyRun" collectionSimCardStatistics = "SimCardDailyStatistics" ) func InitSimCardStatisticsIndexes(ctx context.Context) error { _, err := database.GetCollection(CollectionName_SimCardMonthlyData).Indexes().CreateMany(ctx, []mongo.IndexModel{ {Keys: bson.D{{Key: "month", Value: 1}, {Key: "simId", Value: 1}}}, {Keys: bson.D{{Key: "simId", Value: 1}, {Key: "month", Value: 1}}}, }) return err } type simCardRun struct { SmartHubLastPage *uint64 `bson:"smartHubLastPage"` AcceleronixLastPage *uint64 `bson:"acceleronixLastPage"` SavedPages []string `bson:"savedPages"` Sent bool `bson:"sent"` } func simCardPageKey(page domain.SimCardRunPage) string { return fmt.Sprintf("%s:%d", page.Platform, page.Page) } func MarkSimCardLastPage(date string, platform domain.SimCardPlatform, page uint64) error { field := "smartHubLastPage" if platform == domain.SimPlatformAcceleronix { field = "acceleronixLastPage" } ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() _, err := database.GetCollection(collectionSimCardRun).UpdateOne(ctx, bson.M{"_id": date}, bson.M{"$set": bson.M{field: page}}, options.Update().SetUpsert(true)) return err } func MarkSimCardPagesSaved(date string, pages []domain.SimCardRunPage) error { if len(pages) == 0 { return nil } keys := make([]string, 0, len(pages)) for _, page := range pages { keys = append(keys, simCardPageKey(page)) } ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() _, err := database.GetCollection(collectionSimCardRun).UpdateOne(ctx, bson.M{"_id": date}, bson.M{"$addToSet": bson.M{"savedPages": bson.M{"$each": keys}}}, options.Update().SetUpsert(true)) return err } func IsSimCardRunComplete(date string) (bool, error) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() var run simCardRun err := database.GetCollection(collectionSimCardRun).FindOne(ctx, bson.M{"_id": date}).Decode(&run) if err == mongo.ErrNoDocuments { return false, nil } if err != nil { return false, err } return simCardRunComplete(run), nil } func simCardRunComplete(run simCardRun) bool { if run.Sent || run.SmartHubLastPage == nil || run.AcceleronixLastPage == nil { return false } saved := make(map[string]bool, len(run.SavedPages)) for _, key := range run.SavedPages { saved[key] = true } for page := uint64(0); page <= *run.SmartHubLastPage; page++ { if !saved[simCardPageKey(domain.SimCardRunPage{Platform: domain.SimPlatformSmartHub, Page: page})] { return false } } for page := uint64(1); page <= *run.AcceleronixLastPage; page++ { if !saved[simCardPageKey(domain.SimCardRunPage{Platform: domain.SimPlatformAcceleronix, Page: page})] { return false } } return true } // StreamSimCardDailyUsage returns one snapshot per SIM for the UTC date. // Missing previous-day snapshots are represented by nil and excluded from increments. func StreamSimCardDailyUsage(date time.Time, consume func(total float64, previous *float64) error) error { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) defer cancel() month := date.Format("2006-01") day := date.Day() currentPath := fmt.Sprintf("$dailyUsage.%d.dataUsage", day) match := bson.D{{Key: "$match", Value: bson.M{ "month": bson.M{"$in": []string{month, date.Format("200601")}}, fmt.Sprintf("dailyUsage.%d", day): bson.M{"$exists": true}, }}} pipeline := mongo.Pipeline{ match, {{Key: "$sort", Value: bson.D{{Key: "updateTime", Value: -1}}}}, {{Key: "$group", Value: bson.M{"_id": "$simId", "total": bson.M{"$first": currentPath}, "dailyUsage": bson.M{"$first": "$dailyUsage"}}}}, } previousExpression := any(fmt.Sprintf("$dailyUsage.%d.dataUsage", day-1)) if day == 1 { prev := time.Date(date.Year(), date.Month()-1, 1, 0, 0, 0, 0, time.UTC) prevDay := time.Date(date.Year(), date.Month(), 0, 0, 0, 0, 0, time.UTC).Day() pipeline = append(pipeline, bson.D{{Key: "$lookup", Value: bson.M{ "from": CollectionName_SimCardMonthlyData, "let": bson.M{"id": "$_id"}, "pipeline": mongo.Pipeline{ {{Key: "$match", Value: bson.M{"month": bson.M{"$in": []string{prev.Format("2006-01"), prev.Format("200601")}}, "$expr": bson.M{"$eq": bson.A{"$simId", "$$id"}}}}}, {{Key: "$limit", Value: 1}}, }, "as": "previousMonth", }}}) previousExpression = bson.M{"$arrayElemAt": bson.A{fmt.Sprintf("$previousMonth.dailyUsage.%d.dataUsage", prevDay), 0}} } pipeline = append(pipeline, bson.D{{Key: "$project", Value: bson.M{"total": 1, "previous": previousExpression}}}) cursor, err := database.GetCollection(CollectionName_SimCardMonthlyData).Aggregate(ctx, pipeline, options.Aggregate().SetAllowDiskUse(true)) if err != nil { return err } defer cursor.Close(ctx) for cursor.Next(ctx) { var row struct { Total *float64 `bson:"total"` Previous *float64 `bson:"previous"` } if err := cursor.Decode(&row); err != nil { return err } if row.Total != nil { if err := consume(*row.Total, row.Previous); err != nil { return err } } } return cursor.Err() } func SaveSimCardDailyStatistics(stats domain.SimCardDailyStatistics) error { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() _, err := database.GetCollection(collectionSimCardStatistics).ReplaceOne(ctx, bson.M{"_id": stats.Date}, bson.M{"_id": stats.Date, "date": stats.Date, "cardCount": stats.CardCount, "totalOver200MB": stats.TotalOver200MB, "totalOver1GB": stats.TotalOver1GB, "totalOver2GB": stats.TotalOver2GB, "maxTotalKB": stats.MaxTotalKB, "incrementCardCount": stats.IncrementCardCount, "incrementOver20MB": stats.IncrementOver20MB, "incrementOver50MB": stats.IncrementOver50MB, "incrementOver200MB": stats.IncrementOver200MB, "maxIncrementKB": stats.MaxIncrementKB}, options.Replace().SetUpsert(true)) return err } func MarkSimCardReportSent(date string) error { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() _, err := database.GetCollection(collectionSimCardRun).UpdateOne(ctx, bson.M{"_id": date}, bson.M{"$set": bson.M{"sent": true}}) return err }