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,