mazyar/tm_back
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Enhance dashboard repository and service with caching and aggregation improvements

Browse Source 
- Introduced caching mechanisms for summary and statistics in the dashboard service, improving performance by reducing redundant data retrieval.
- Refactored the Summary method to utilize MongoDB aggregation for more efficient data processing and retrieval.
- Added synchronization features using singleflight to prevent duplicate processing of requests for cached data.
- Updated the repository to include a cachedScrapedTendersScope method, enhancing the efficiency of scraped document statistics retrieval.

This update significantly optimizes the dashboard's performance and data handling capabilities, ensuring faster response times and reduced load on the database.
This commit is contained in:
Mazyar
2026-06-21 09:36:12 +03:30
parent 6fb57c41c1
commit dfab3e17d2
4 changed files with 260 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/dashboard/repository.go
+97 -111
View File
@@ -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) {
internal/dashboard/service.go
+95 -30
View File
@@ -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,
}
internal/dashboard/statistics_repository.go
+58 -1
View File
@@ -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 {
internal/dashboard/statistics_repository_test.go
+10
View File
@@ -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)
}
}