From 8dbd9927b08a4dd7a26cdc27f194fd9c2c81ba02 Mon Sep 17 00:00:00 2001 From: Mazyar Date: Tue, 30 Jun 2026 19:03:22 +0330 Subject: [PATCH] Enhance dashboard service with Redis caching for statistics - Updated the dashboard service to integrate Redis caching for improved performance in statistics retrieval. - Modified the NewService function to accept a Redis client, enabling caching of dashboard statistics. - Implemented logic to retrieve statistics from Redis, falling back to the database if necessary, and introduced a background process to warm the cache. - Enhanced error handling and logging for Redis operations to ensure robust statistics management. - Increased cache duration for scraped documents and adjusted timeout settings for MongoDB queries to optimize performance. This update significantly improves the responsiveness and efficiency of the dashboard by leveraging Redis for caching statistics. --- cmd/web/main.go | 2 +- internal/dashboard/service.go | 121 +++++++++++++- internal/dashboard/statistics_repository.go | 172 ++++++++++++-------- 3 files changed, 218 insertions(+), 77 deletions(-) diff --git a/cmd/web/main.go b/cmd/web/main.go index 382b936..81fa6ad 100644 --- a/cmd/web/main.go +++ b/cmd/web/main.go @@ -253,7 +253,7 @@ func main() { kanbanService := kanban.NewService(boardRepository, columnRepository, cardRepository, goRulesClient, conf.GoRules.RuleID, logger) documentScraperService := document_scraper.NewService(tenderRepository, aiPipelineClient, logger) dashboardRepository := dashboard.NewRepository(mongoManager, logger, procedureDocumentsLister) - dashboardService := dashboard.NewService(dashboardRepository, logger) + dashboardService := dashboard.NewService(dashboardRepository, logger, redisClient) aiPipelineService := ai_pipeline.NewService(aiPipelineClient, logger, tenderService) auditLogRepository := auditlog.NewRepository(auditStore) auditLogService := auditlog.NewService(auditLogRepository, logger) diff --git a/internal/dashboard/service.go b/internal/dashboard/service.go index 102a3e5..b8b462c 100644 --- a/internal/dashboard/service.go +++ b/internal/dashboard/service.go @@ -2,13 +2,16 @@ package dashboard import ( "context" + "encoding/json" "fmt" "strconv" "strings" "sync" "time" "tm/pkg/logger" + "tm/pkg/redis" + goredis "github.com/redis/go-redis/v9" "golang.org/x/sync/singleflight" ) @@ -44,6 +47,7 @@ type cacheEntry[T any] struct { type service struct { repo Repository logger logger.Logger + redis redis.Client summaryMu sync.Mutex summary map[int64]cacheEntry[*SummaryResponse] summaryGroup singleflight.Group @@ -53,13 +57,16 @@ type service struct { } // NewService creates a dashboard service. -func NewService(repo Repository, log logger.Logger) Service { - return &service{ +func NewService(repo Repository, log logger.Logger, redisClient redis.Client) Service { + s := &service{ repo: repo, logger: log, + redis: redisClient, summary: make(map[int64]cacheEntry[*SummaryResponse]), statistics: make(map[int]cacheEntry[*StatisticsReportResponse]), } + go s.warmStatisticsCache() + return s } func (s *service) Summary(ctx context.Context, query SummaryQuery) (*SummaryResponse, error) { @@ -210,11 +217,60 @@ func (s *service) Statistics(ctx context.Context, query StatisticsQuery) (*Stati return stale, nil } - return s.loadStatistics(ctx, days) + if redisCached, ok := s.getRedisStatistics(ctx, days); ok { + s.storeStatistics(days, redisCached) + go s.refreshStatistics(days) + return redisCached, nil + } + + result, err := s.loadStatistics(ctx, days) + if err == nil { + return result, nil + } + + if redisCached, ok := s.getRedisStatistics(ctx, days); ok { + s.storeStatistics(days, redisCached) + return redisCached, nil + } + + s.logger.Error("Serving empty dashboard statistics after fetch failure", map[string]interface{}{ + "days": days, + "error": err.Error(), + }) + return emptyStatisticsReport(days), nil +} + +func (s *service) warmStatisticsCache() { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + if cached, ok := s.getRedisStatistics(ctx, defaultTrendDays); ok { + s.storeStatistics(defaultTrendDays, cached) + s.logger.Info("Dashboard statistics cache warmed from Redis", map[string]interface{}{ + "days": defaultTrendDays, + }) + return + } + + go s.refreshStatistics(defaultTrendDays) +} + +func emptyStatisticsReport(days int) *StatisticsReportResponse { + return &StatisticsReportResponse{ + Days: days, + GeneratedAt: time.Now().UTC().Unix(), + Daily: StatisticsDailySeries{ + ScrapedTED: fillTrendSeries(days, nil), + ScrapedDocuments: fillTrendSeries(days, nil), + TranslatedNotices: fillTrendSeries(days, nil), + }, + } } func (s *service) refreshStatistics(days int) { - _, _ = s.loadStatistics(context.Background(), days) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + _, _ = s.loadStatistics(ctx, days) } func (s *service) loadStatistics(ctx context.Context, days int) (*StatisticsReportResponse, error) { @@ -239,6 +295,7 @@ func (s *service) loadStatistics(ctx context.Context, days int) (*StatisticsRepo } s.storeStatistics(days, result) + s.storeRedisStatistics(context.Background(), days, result) return result, nil }) if err != nil { @@ -320,6 +377,62 @@ func (s *service) storeStatistics(days int, value *StatisticsReportResponse) { } } +func statisticsRedisKey(days int) string { + return fmt.Sprintf("dashboard:statistics:%d", days) +} + +func (s *service) getRedisStatistics(ctx context.Context, days int) (*StatisticsReportResponse, bool) { + if s.redis == nil { + return nil, false + } + + raw, err := s.redis.Get(ctx, statisticsRedisKey(days)) + if err != nil { + if err != goredis.Nil { + s.logger.Warn("Failed to read dashboard statistics cache from Redis", map[string]interface{}{ + "days": days, + "error": err.Error(), + }) + } + return nil, false + } + + var report StatisticsReportResponse + if err := json.Unmarshal([]byte(raw), &report); err != nil { + s.logger.Warn("Failed to decode dashboard statistics cache from Redis", map[string]interface{}{ + "days": days, + "error": err.Error(), + }) + _ = s.redis.Del(ctx, statisticsRedisKey(days)) + return nil, false + } + + return &report, true +} + +func (s *service) storeRedisStatistics(ctx context.Context, days int, value *StatisticsReportResponse) { + if s.redis == nil || value == nil { + return + } + + encoded, err := json.Marshal(value) + if err != nil { + s.logger.Warn("Failed to encode dashboard statistics for Redis", map[string]interface{}{ + "days": days, + "error": err.Error(), + }) + return + } + + ttl := statisticsCacheTTL + statisticsStaleGraceTTL + if err := s.redis.Set(ctx, statisticsRedisKey(days), string(encoded), ttl); err != nil { + s.logger.Warn("Failed to store dashboard statistics in Redis", map[string]interface{}{ + "days": days, + "error": err.Error(), + }) + } +} + func closingWindowHours(hours int) int { if hours <= 0 { return defaultClosingWindowHours diff --git a/internal/dashboard/statistics_repository.go b/internal/dashboard/statistics_repository.go index 23997f8..8ba1d4f 100644 --- a/internal/dashboard/statistics_repository.go +++ b/internal/dashboard/statistics_repository.go @@ -2,8 +2,8 @@ package dashboard import ( "context" - "fmt" "strings" + "sync" "time" "tm/internal/notice" @@ -12,19 +12,18 @@ import ( "go.mongodb.org/mongo-driver/v2/bson" mongodriver "go.mongodb.org/mongo-driver/v2/mongo" - "golang.org/x/sync/errgroup" ) 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 +const scrapedScopeCacheTTL = 15 * time.Minute const ( - scrapedScopeResolveTimeout = 2 * time.Minute - statisticsQueryTimeout = 2 * time.Minute - maxScrapedFolderInClause = 5000 + scrapedScopeResolveTimeout = 3 * time.Minute + statisticsMongoQueryTimeout = 90 * time.Second + maxScrapedFolderInClause = 5000 ) // scrapedTendersScope is the resolved Mongo filter for tenders with scraped documents. @@ -38,8 +37,7 @@ type scrapedTendersScope struct { } func (r *repository) Statistics(ctx context.Context, days int, startUnix int64) (*StatisticsReportResponse, error) { - queryCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), statisticsQueryTimeout) - defer cancel() + _ = ctx now := time.Now().UTC() endDay := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC) @@ -54,77 +52,109 @@ func (r *repository) Statistics(ctx context.Context, days int, startUnix int64) scrapedScope scrapedTendersScope ) - g, gctx := errgroup.WithContext(queryCtx) + var wg sync.WaitGroup + var mu sync.Mutex - g.Go(func() error { - // Use created_at (first ingest time). processing_metadata.scraped_at is reset on - // notice refresh merges, which would collapse historical scrapes onto the latest day. - counts, err := r.dailyCountByTimestamp(gctx, noticesCollectionName, bson.M{ + recordFailure := func(section string, err error) { + mu.Lock() + defer mu.Unlock() + r.logger.Warn("Dashboard statistics section failed", map[string]interface{}{ + "section": section, + "error": err.Error(), + }) + } + + wg.Add(1) + go func() { + defer wg.Done() + qctx, cancel := context.WithTimeout(context.Background(), statisticsMongoQueryTimeout) + defer cancel() + + counts, err := r.dailyCountByTimestamp(qctx, noticesCollectionName, bson.M{ "source": notice.TenderSourceTEDScraper, "created_at": bson.M{"$gte": startUnix, "$gt": 0}, }, "created_at") if err != nil { - return fmt.Errorf("scraped ted per day: %w", err) + recordFailure("scraped_ted", err) + scrapedTED = map[string]int64{} + return } scrapedTED = counts - return nil - }) + }() - g.Go(func() error { - counts, err := r.translatedNoticesPerDay(gctx, startDay, endDay) + wg.Add(1) + go func() { + defer wg.Done() + qctx, cancel := context.WithTimeout(context.Background(), statisticsMongoQueryTimeout) + defer cancel() + + counts, err := r.translatedNoticesPerDay(qctx, startDay, endDay) if err != nil { - return fmt.Errorf("translated notices per day: %w", err) + recordFailure("translated_notices", err) + translatedNotices = map[string]int64{} + return } translatedNotices = counts - return nil - }) + }() - g.Go(func() error { - total, err := r.metricsCounter.Get(gctx, orm.AITranslationSuccessCounterKey) + wg.Add(1) + go func() { + defer wg.Done() + qctx, cancel := context.WithTimeout(context.Background(), statisticsMongoQueryTimeout) + defer cancel() + + total, err := r.metricsCounter.Get(qctx, orm.AITranslationSuccessCounterKey) if err != nil { - return fmt.Errorf("total translated tenders: %w", err) + recordFailure("translated_total", err) + return } totalTranslated = total - return nil - }) + }() - // Resolve scraped-document scope from MinIO (cached) in parallel with Mongo queries. - g.Go(func() error { - scope, err := r.cachedScrapedTendersScope(gctx) + wg.Add(1) + go func() { + defer wg.Done() + scope, err := r.cachedScrapedTendersScope(context.Background()) if err != nil { - return fmt.Errorf("scraped tenders scope: %w", err) + recordFailure("scraped_scope", err) + scrapedScope = mongoScrapedTendersScope() + return } scrapedScope = scope - return nil - }) + }() - if err := g.Wait(); err != nil { - return nil, err - } + wg.Wait() - g2, gctx2 := errgroup.WithContext(queryCtx) + wg.Add(1) + go func() { + defer wg.Done() + qctx, cancel := context.WithTimeout(context.Background(), statisticsMongoQueryTimeout) + defer cancel() - g2.Go(func() error { - counts, err := r.scrapedDocumentsPerDay(gctx2, startUnix, scrapedScope) + counts, err := r.scrapedDocumentsPerDay(qctx, startUnix, scrapedScope) if err != nil { - return fmt.Errorf("scraped documents per day: %w", err) + recordFailure("scraped_documents", err) + scrapedDocuments = map[string]int64{} + return } scrapedDocuments = counts - return nil - }) + }() - g2.Go(func() error { - total, err := r.totalScrapedTenders(gctx2, scrapedScope) + wg.Add(1) + go func() { + defer wg.Done() + qctx, cancel := context.WithTimeout(context.Background(), statisticsMongoQueryTimeout) + defer cancel() + + total, err := r.totalScrapedTenders(qctx, scrapedScope) if err != nil { - return fmt.Errorf("total scraped tenders: %w", err) + recordFailure("scraped_documents_total", err) + return } totalDocuments = total - return nil - }) + }() - if err := g2.Wait(); err != nil { - return nil, err - } + wg.Wait() return &StatisticsReportResponse{ Days: days, @@ -195,23 +225,17 @@ func filterDailyCountsSince(counts map[string]int64, startUnix int64) map[string } func (r *repository) scrapedDocumentsPerDayFromMongo(ctx context.Context, startUnix int64, match bson.M) (map[string]int64, error) { + combinedMatch := bson.M{ + "processing_metadata.documents_scraped_at": bson.M{"$gte": startUnix, "$gt": 0}, + } + for key, value := range match { + combinedMatch[key] = value + } + pipeline := mongodriver.Pipeline{ - {{Key: "$match", Value: match}}, - {{Key: "$unwind", Value: "$scraped_documents"}}, + {{Key: "$match", Value: combinedMatch}}, {{Key: "$addFields", Value: bson.M{ - "metric_ts": bson.M{ - "$cond": bson.A{ - bson.M{"$gt": bson.A{"$scraped_documents.scraped_at", 0}}, - "$scraped_documents.scraped_at", - "$processing_metadata.documents_scraped_at", - }, - }, - }}}, - {{Key: "$addFields", Value: bson.M{ - "metric_ts": normalizeTimestampExpr("metric_ts"), - }}}, - {{Key: "$match", Value: bson.M{ - "metric_ts": bson.M{"$gte": startUnix, "$gt": 0}, + "metric_ts": normalizeTimestampExpr("processing_metadata.documents_scraped_at"), }}}, {{Key: "$group", Value: bson.M{ "_id": bson.M{ @@ -258,7 +282,7 @@ func mongoScrapedTendersScope() scrapedTendersScope { } } -func (r *repository) cachedScrapedTendersScope(ctx context.Context) (scrapedTendersScope, error) { +func (r *repository) cachedScrapedTendersScope(_ context.Context) (scrapedTendersScope, error) { now := time.Now() r.scrapedScopeMu.Lock() @@ -283,9 +307,11 @@ func (r *repository) cachedScrapedTendersScope(ctx context.Context) (scrapedTend scope, resolveErr := r.resolveScrapedTendersScope(resolveCtx) if resolveErr != nil { - r.logger.Warn("MinIO scraped scope unavailable, using Mongo fallback for dashboard statistics", map[string]interface{}{ - "error": resolveErr.Error(), - }) + if r.logger != nil { + r.logger.Warn("MinIO scraped scope unavailable, using Mongo fallback for dashboard statistics", map[string]interface{}{ + "error": resolveErr.Error(), + }) + } return mongoScrapedTendersScope(), nil } @@ -314,10 +340,12 @@ func (r *repository) resolveScrapedTendersScope(ctx context.Context) (scrapedTen if len(folderIDs) > 0 { totalTenders := int64(len(folderIDs)) if len(folderIDs) > maxScrapedFolderInClause { - r.logger.Warn("Too many MinIO procedure folders for dashboard statistics, using MinIO counts without Mongo $in filter", map[string]interface{}{ - "folder_count": len(folderIDs), - "max_in_clause": maxScrapedFolderInClause, - }) + if r.logger != nil { + r.logger.Warn("Too many MinIO procedure folders for dashboard statistics, using MinIO counts without Mongo $in filter", map[string]interface{}{ + "folder_count": len(folderIDs), + "max_in_clause": maxScrapedFolderInClause, + }) + } return scrapedTendersScope{ fromMinIO: true, dailyCounts: dailyCounts,