diff --git a/internal/dashboard/repository.go b/internal/dashboard/repository.go index d019b53..0b05df0 100644 --- a/internal/dashboard/repository.go +++ b/internal/dashboard/repository.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "strings" + "sync" "time" "tm/internal/tender" "tm/pkg/ai_summarizer" @@ -12,6 +13,7 @@ import ( "go.mongodb.org/mongo-driver/v2/bson" "go.mongodb.org/mongo-driver/v2/mongo" + "golang.org/x/sync/singleflight" ) // ProcedureDocumentsLister lists contract folders that have scraped documents in MinIO. @@ -34,11 +36,16 @@ type Repository interface { } type repository struct { - ormRepo orm.Repository[tender.Tender] - mongoManager *orm.ConnectionManager - metricsCounter *orm.Counter - procedureLister ProcedureDocumentsLister - logger logger.Logger + ormRepo orm.Repository[tender.Tender] + mongoManager *orm.ConnectionManager + metricsCounter *orm.Counter + procedureLister ProcedureDocumentsLister + logger logger.Logger + scrapedScopeMu sync.Mutex + scrapedScope scrapedTendersScope + scrapedScopeExpiry time.Time + scrapedScopeCached bool + scrapedScopeGroup singleflight.Group } // NewRepository creates a dashboard repository backed by the tenders collection. @@ -60,135 +67,114 @@ func (r *repository) Summary(ctx context.Context, closingWindowSec int64) (*Summ now := time.Now().Unix() windowEnd := now + closingWindowSec - total, err := r.ormRepo.Count(ctx, bson.M{}) - if err != nil { - return nil, fmt.Errorf("count total tenders: %w", err) + pipeline := mongo.Pipeline{ + {{Key: "$facet", Value: bson.M{ + "total": bson.A{ + bson.M{"$count": "count"}, + }, + "active": bson.A{ + bson.M{"$match": bson.M{"status": tender.TenderStatusActive}}, + bson.M{"$count": "count"}, + }, + "awarded": bson.A{ + bson.M{"$match": bson.M{"status": tender.TenderStatusAwarded}}, + bson.M{"$count": "count"}, + }, + "cancelled": bson.A{ + bson.M{"$match": bson.M{"status": tender.TenderStatusCancelled}}, + bson.M{"$count": "count"}, + }, + "expired": bson.A{ + bson.M{"$addFields": bson.M{"expiry_deadline": expiryDeadlineExpr()}}, + bson.M{"$match": bson.M{ + "status": bson.M{"$nin": []tender.TenderStatus{ + tender.TenderStatusAwarded, + tender.TenderStatusCancelled, + }}, + "$or": []bson.M{ + {"status": tender.TenderStatusExpired}, + {"expiry_deadline": bson.M{"$gt": 0, "$lte": now}}, + }, + }}, + bson.M{"$count": "count"}, + }, + "closing_soon": bson.A{ + bson.M{"$addFields": bson.M{"effective_deadline": effectiveDeadlineExpr()}}, + bson.M{"$match": bson.M{ + "effective_deadline": bson.M{"$gt": now, "$lte": windowEnd}, + }}, + bson.M{"$count": "count"}, + }, + "currency_value": bson.A{ + bson.M{"$match": bson.M{ + "estimated_value": bson.M{"$gt": 0}, + "currency": bson.M{"$type": "string", "$ne": ""}, + }}, + bson.M{"$group": bson.M{ + "_id": "$currency", + "count": bson.M{"$sum": 1}, + "total": bson.M{"$sum": "$estimated_value"}, + }}, + bson.M{"$sort": bson.M{"count": -1}}, + bson.M{"$limit": 1}, + }, + }}}, } - active, err := r.ormRepo.Count(ctx, bson.M{"status": tender.TenderStatusActive}) + results, err := r.ormRepo.Aggregate(ctx, pipeline) if err != nil { - return nil, fmt.Errorf("count active tenders: %w", err) + return nil, fmt.Errorf("dashboard summary aggregation: %w", err) + } + if len(results) == 0 { + return &SummaryResponse{ + ValueCurrency: defaultValueCurrency, + GeneratedAt: now, + }, nil } - awarded, err := r.ormRepo.Count(ctx, bson.M{"status": tender.TenderStatusAwarded}) - if err != nil { - return nil, fmt.Errorf("count awarded tenders: %w", err) - } - - expired, err := r.countExpired(ctx, now) - if err != nil { - return nil, err - } - - cancelled, err := r.ormRepo.Count(ctx, bson.M{"status": tender.TenderStatusCancelled}) - if err != nil { - return nil, fmt.Errorf("count cancelled tenders: %w", err) - } - - closingSoon, err := r.countClosingSoon(ctx, now, windowEnd) - if err != nil { - return nil, err - } - - valueCurrency, totalValue, err := r.dominantCurrencyValue(ctx) - if err != nil { - return nil, err - } + facet := results[0] + valueCurrency, totalValue := facetCurrencyValue(facet) return &SummaryResponse{ - TotalTenders: total, - ActiveTenders: active, - AwardedTenders: awarded, - ExpiredTenders: expired, - CancelledTenders: cancelled, - ClosingSoon: closingSoon, + TotalTenders: facetCount(facet, "total"), + ActiveTenders: facetCount(facet, "active"), + AwardedTenders: facetCount(facet, "awarded"), + ExpiredTenders: facetCount(facet, "expired"), + CancelledTenders: facetCount(facet, "cancelled"), + ClosingSoon: facetCount(facet, "closing_soon"), TotalEstimatedValue: totalValue, ValueCurrency: valueCurrency, GeneratedAt: now, }, nil } -func (r *repository) countExpired(ctx context.Context, now int64) (int64, error) { - pipeline := mongo.Pipeline{ - {{Key: "$addFields", Value: bson.M{"expiry_deadline": expiryDeadlineExpr()}}}, - {{Key: "$match", Value: bson.M{ - "status": bson.M{"$nin": []tender.TenderStatus{ - tender.TenderStatusAwarded, - tender.TenderStatusCancelled, - }}, - "$or": []bson.M{ - {"status": tender.TenderStatusExpired}, - { - "expiry_deadline": bson.M{"$gt": 0, "$lte": now}, - }, - }, - }}}, - {{Key: "$count", Value: "count"}}, +func facetCount(facet bson.M, key string) int64 { + rows, ok := facet[key].(bson.A) + if !ok || len(rows) == 0 { + return 0 } - - results, err := r.ormRepo.Aggregate(ctx, pipeline) - if err != nil { - return 0, fmt.Errorf("count expired tenders: %w", err) + row, ok := rows[0].(bson.M) + if !ok { + return 0 } - - if len(results) == 0 { - return 0, nil - } - - return aggregateCount(results[0], "count"), nil + return aggregateCount(row, "count") } -func (r *repository) countClosingSoon(ctx context.Context, now, windowEnd int64) (int64, error) { - pipeline := mongo.Pipeline{ - {{Key: "$addFields", Value: bson.M{"effective_deadline": effectiveDeadlineExpr()}}}, - {{Key: "$match", Value: bson.M{ - "effective_deadline": bson.M{"$gt": now, "$lte": windowEnd}, - }}}, - {{Key: "$count", Value: "count"}}, +func facetCurrencyValue(facet bson.M) (string, float64) { + rows, ok := facet["currency_value"].(bson.A) + if !ok || len(rows) == 0 { + return defaultValueCurrency, 0 } - - results, err := r.ormRepo.Aggregate(ctx, pipeline) - if err != nil { - return 0, fmt.Errorf("count closing soon: %w", err) - } - - if len(results) == 0 { - return 0, nil - } - - return aggregateCount(results[0], "count"), nil -} - -func (r *repository) dominantCurrencyValue(ctx context.Context) (string, float64, error) { - pipeline := mongo.Pipeline{ - {{Key: "$match", Value: bson.M{ - "estimated_value": bson.M{"$gt": 0}, - "currency": bson.M{"$type": "string", "$ne": ""}, - }}}, - {{Key: "$group", Value: bson.M{ - "_id": "$currency", - "count": bson.M{"$sum": 1}, - "total": bson.M{"$sum": "$estimated_value"}, - }}}, - {{Key: "$sort", Value: bson.M{"count": -1}}}, - {{Key: "$limit", Value: 1}}, - } - - results, err := r.ormRepo.Aggregate(ctx, pipeline) - if err != nil { - return defaultValueCurrency, 0, fmt.Errorf("dominant currency value: %w", err) - } - - if len(results) == 0 { - return defaultValueCurrency, 0, nil + row, ok := rows[0].(bson.M) + if !ok { + return defaultValueCurrency, 0 } currency := defaultValueCurrency - if c, ok := results[0]["_id"].(string); ok && c != "" { + if c, ok := row["_id"].(string); ok && c != "" { currency = strings.ToUpper(c) } - - return currency, toFloat64(results[0]["total"]), nil + return currency, toFloat64(row["total"]) } func (r *repository) Trend(ctx context.Context, days int, metric string, startUnix int64) (map[string]int64, error) { diff --git a/internal/dashboard/service.go b/internal/dashboard/service.go index 7e8a008..0dc5bf6 100644 --- a/internal/dashboard/service.go +++ b/internal/dashboard/service.go @@ -3,10 +3,13 @@ package dashboard import ( "context" "fmt" + "strconv" "strings" "sync" "time" "tm/pkg/logger" + + "golang.org/x/sync/singleflight" ) const ( @@ -16,6 +19,7 @@ const ( defaultCountriesLimit = 6 defaultListLimit = 5 maxListLimit = 20 + summaryCacheTTL = 60 * time.Second statisticsCacheTTL = 60 * time.Second ) @@ -30,16 +34,20 @@ type Service interface { Statistics(ctx context.Context, query StatisticsQuery) (*StatisticsReportResponse, error) } -type statisticsCacheEntry struct { +type cacheEntry[T any] struct { expiresAt time.Time - value *StatisticsReportResponse + value T } type service struct { - repo Repository - logger logger.Logger - statisticsMu sync.Mutex - statistics map[int]statisticsCacheEntry + repo Repository + logger logger.Logger + summaryMu sync.Mutex + summary map[int64]cacheEntry[*SummaryResponse] + summaryGroup singleflight.Group + statisticsMu sync.Mutex + statistics map[int]cacheEntry[*StatisticsReportResponse] + statisticsGroup singleflight.Group } // NewService creates a dashboard service. @@ -47,7 +55,8 @@ func NewService(repo Repository, log logger.Logger) Service { return &service{ repo: repo, logger: log, - statistics: make(map[int]statisticsCacheEntry), + summary: make(map[int64]cacheEntry[*SummaryResponse]), + statistics: make(map[int]cacheEntry[*StatisticsReportResponse]), } } @@ -55,19 +64,36 @@ func (s *service) Summary(ctx context.Context, query SummaryQuery) (*SummaryResp windowHours := closingWindowHours(query.ClosingWindow) windowSec := int64(windowHours) * 3600 - s.logger.Info("Fetching dashboard summary", map[string]interface{}{ - "closing_window_hours": windowHours, - }) - - out, err := s.repo.Summary(ctx, windowSec) - if err != nil { - s.logger.Error("Failed to fetch dashboard summary", map[string]interface{}{ - "error": err.Error(), - }) - return nil, fmt.Errorf("dashboard summary: %w", err) + if cached, ok := s.cachedSummary(windowSec); ok { + return cached, nil } - return out, nil + cacheKey := strconv.FormatInt(windowSec, 10) + out, err, _ := s.summaryGroup.Do(cacheKey, func() (interface{}, error) { + if cached, ok := s.cachedSummary(windowSec); ok { + return cached, nil + } + + s.logger.Info("Fetching dashboard summary", map[string]interface{}{ + "closing_window_hours": windowHours, + }) + + result, err := s.repo.Summary(ctx, windowSec) + if err != nil { + s.logger.Error("Failed to fetch dashboard summary", map[string]interface{}{ + "error": err.Error(), + }) + return nil, fmt.Errorf("dashboard summary: %w", err) + } + + s.storeSummary(windowSec, result) + return result, nil + }) + if err != nil { + return nil, err + } + + return out.(*SummaryResponse), nil } func (s *service) Trend(ctx context.Context, query TrendQuery) (*TrendResponse, error) { @@ -177,22 +203,61 @@ func (s *service) Statistics(ctx context.Context, query StatisticsQuery) (*Stati return cached, nil } - startUnix := trendStartUnix(days) + cacheKey := strconv.Itoa(days) + out, err, _ := s.statisticsGroup.Do(cacheKey, func() (interface{}, error) { + if cached, ok := s.cachedStatistics(days); ok { + return cached, nil + } - s.logger.Info("Fetching dashboard statistics report", map[string]interface{}{ - "days": days, - }) + startUnix := trendStartUnix(days) - out, err := s.repo.Statistics(ctx, days, startUnix) - if err != nil { - s.logger.Error("Failed to fetch dashboard statistics report", map[string]interface{}{ - "error": err.Error(), + s.logger.Info("Fetching dashboard statistics report", map[string]interface{}{ + "days": days, }) - return nil, fmt.Errorf("dashboard statistics: %w", err) + + result, err := s.repo.Statistics(ctx, days, startUnix) + if err != nil { + s.logger.Error("Failed to fetch dashboard statistics report", map[string]interface{}{ + "error": err.Error(), + }) + return nil, fmt.Errorf("dashboard statistics: %w", err) + } + + s.storeStatistics(days, result) + return result, nil + }) + if err != nil { + return nil, err } - s.storeStatistics(days, out) - return out, nil + return out.(*StatisticsReportResponse), nil +} + +func (s *service) cachedSummary(windowSec int64) (*SummaryResponse, bool) { + now := time.Now() + + s.summaryMu.Lock() + defer s.summaryMu.Unlock() + + entry, ok := s.summary[windowSec] + if !ok || now.After(entry.expiresAt) { + if ok { + delete(s.summary, windowSec) + } + return nil, false + } + + return entry.value, true +} + +func (s *service) storeSummary(windowSec int64, value *SummaryResponse) { + s.summaryMu.Lock() + defer s.summaryMu.Unlock() + + s.summary[windowSec] = cacheEntry[*SummaryResponse]{ + expiresAt: time.Now().Add(summaryCacheTTL), + value: value, + } } func (s *service) cachedStatistics(days int) (*StatisticsReportResponse, bool) { @@ -216,7 +281,7 @@ func (s *service) storeStatistics(days int, value *StatisticsReportResponse) { s.statisticsMu.Lock() defer s.statisticsMu.Unlock() - s.statistics[days] = statisticsCacheEntry{ + s.statistics[days] = cacheEntry[*StatisticsReportResponse]{ expiresAt: time.Now().Add(statisticsCacheTTL), value: value, } diff --git a/internal/dashboard/statistics_repository.go b/internal/dashboard/statistics_repository.go index a63cb42..c03215f 100644 --- a/internal/dashboard/statistics_repository.go +++ b/internal/dashboard/statistics_repository.go @@ -16,6 +16,10 @@ import ( const noticesCollectionName = "notices" +// scrapedScopeCacheTTL avoids listing every MinIO object on each statistics request while +// keeping counts aligned with the tender panel (MinIO is the source of truth for scraped docs). +const scrapedScopeCacheTTL = 5 * time.Minute + // scrapedTendersScope is the resolved Mongo filter for tenders with scraped documents. // zero is set when MinIO reports no procedure folders; callers must return zero without querying. type scrapedTendersScope struct { @@ -28,7 +32,8 @@ func (r *repository) Statistics(ctx context.Context, days int, startUnix int64) endDay := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC) startDay := endDay.AddDate(0, 0, -(days - 1)) - scrapedScope, err := r.resolveScrapedTendersScope(ctx) + // Resolve scraped-document scope from MinIO (cached); falls back to Mongo flags on error. + scrapedScope, err := r.cachedScrapedTendersScope(ctx) if err != nil { return nil, err } @@ -187,6 +192,58 @@ func (r *repository) totalScrapedTenders(ctx context.Context, scope scrapedTende return r.ormRepo.Count(ctx, scope.match) } +func mongoScrapedTendersScope() scrapedTendersScope { + return scrapedTendersScope{ + match: bson.M{ + "processing_metadata.documents_scraped": true, + "scraped_documents": bson.M{"$exists": true, "$ne": bson.A{}}, + }, + } +} + +func (r *repository) cachedScrapedTendersScope(ctx context.Context) (scrapedTendersScope, error) { + now := time.Now() + + r.scrapedScopeMu.Lock() + if r.scrapedScopeCached && now.Before(r.scrapedScopeExpiry) { + scope := r.scrapedScope + r.scrapedScopeMu.Unlock() + return scope, nil + } + r.scrapedScopeMu.Unlock() + + v, err, _ := r.scrapedScopeGroup.Do("scraped-scope", func() (interface{}, error) { + r.scrapedScopeMu.Lock() + if r.scrapedScopeCached && time.Now().Before(r.scrapedScopeExpiry) { + scope := r.scrapedScope + r.scrapedScopeMu.Unlock() + return scope, nil + } + r.scrapedScopeMu.Unlock() + + scope, resolveErr := r.resolveScrapedTendersScope(ctx) + if resolveErr != nil { + r.logger.Warn("MinIO scraped scope unavailable, using Mongo fallback for dashboard statistics", map[string]interface{}{ + "error": resolveErr.Error(), + }) + return mongoScrapedTendersScope(), nil + } + + r.scrapedScopeMu.Lock() + r.scrapedScope = scope + r.scrapedScopeExpiry = time.Now().Add(scrapedScopeCacheTTL) + r.scrapedScopeCached = true + r.scrapedScopeMu.Unlock() + + return scope, nil + }) + if err != nil { + return scrapedTendersScope{}, err + } + + return v.(scrapedTendersScope), nil +} + func (r *repository) resolveScrapedTendersScope(ctx context.Context) (scrapedTendersScope, error) { folderIDs, err := r.scrapedDocumentsFolderIDs(ctx) if err != nil { diff --git a/internal/dashboard/statistics_repository_test.go b/internal/dashboard/statistics_repository_test.go index ac9302f..c30b715 100644 --- a/internal/dashboard/statistics_repository_test.go +++ b/internal/dashboard/statistics_repository_test.go @@ -129,3 +129,13 @@ func TestResolveScrapedTendersScopeMongoFallbackWithoutLister(t *testing.T) { t.Fatalf("expected mongo fallback filter, got %#v", scope.match) } } + +func TestMongoScrapedTendersScopeMatchesStatisticsFilter(t *testing.T) { + scope := mongoScrapedTendersScope() + if scope.zero { + t.Fatal("expected non-zero mongo statistics scope") + } + if scope.match["processing_metadata.documents_scraped"] != true { + t.Fatalf("expected documents_scraped filter, got %#v", scope.match) + } +}