diff --git a/cmd/web/bootstrap/config.go b/cmd/web/bootstrap/config.go index 0002db2..237f953 100644 --- a/cmd/web/bootstrap/config.go +++ b/cmd/web/bootstrap/config.go @@ -76,7 +76,7 @@ type AISummarizerConfig struct { MinioSecretKey string `env:"AI_SUMMARIZER_MINIO_SECRET_KEY" envDefault:""` MinioUseSSL bool `env:"AI_SUMMARIZER_MINIO_USE_SSL" envDefault:"false"` MinioRegion string `env:"AI_SUMMARIZER_MINIO_REGION" envDefault:"us-east-1"` - MinioBucket string `env:"AI_SUMMARIZER_MINIO_BUCKET" envDefault:"opplens-documents"` + MinioBucket string `env:"AI_SUMMARIZER_MINIO_BUCKET" envDefault:"opplens"` } // GoRulesConfig holds configuration for the GoRules status engine. diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index 31c4422..35c3a24 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -48,5 +48,5 @@ type AISummarizerConfig struct { MinioSecretKey string `env:"AI_SUMMARIZER_MINIO_SECRET_KEY" envDefault:""` MinioUseSSL bool `env:"AI_SUMMARIZER_MINIO_USE_SSL" envDefault:"false"` MinioRegion string `env:"AI_SUMMARIZER_MINIO_REGION" envDefault:"us-east-1"` - MinioBucket string `env:"AI_SUMMARIZER_MINIO_BUCKET" envDefault:"opplens-documents"` + MinioBucket string `env:"AI_SUMMARIZER_MINIO_BUCKET" envDefault:"opplens"` } diff --git a/internal/dashboard/service.go b/internal/dashboard/service.go index 73b5ecd..7e8a008 100644 --- a/internal/dashboard/service.go +++ b/internal/dashboard/service.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "strings" + "sync" "time" "tm/pkg/logger" ) @@ -15,6 +16,7 @@ const ( defaultCountriesLimit = 6 defaultListLimit = 5 maxListLimit = 20 + statisticsCacheTTL = 60 * time.Second ) // Service defines dashboard business operations. @@ -28,14 +30,25 @@ type Service interface { Statistics(ctx context.Context, query StatisticsQuery) (*StatisticsReportResponse, error) } +type statisticsCacheEntry struct { + expiresAt time.Time + value *StatisticsReportResponse +} + type service struct { - repo Repository - logger logger.Logger + repo Repository + logger logger.Logger + statisticsMu sync.Mutex + statistics map[int]statisticsCacheEntry } // NewService creates a dashboard service. func NewService(repo Repository, log logger.Logger) Service { - return &service{repo: repo, logger: log} + return &service{ + repo: repo, + logger: log, + statistics: make(map[int]statisticsCacheEntry), + } } func (s *service) Summary(ctx context.Context, query SummaryQuery) (*SummaryResponse, error) { @@ -160,6 +173,10 @@ func (s *service) Recent(ctx context.Context, query RecentQuery) (*RecentRespons func (s *service) Statistics(ctx context.Context, query StatisticsQuery) (*StatisticsReportResponse, error) { days := trendDays(query.Days) + if cached, ok := s.cachedStatistics(days); ok { + return cached, nil + } + startUnix := trendStartUnix(days) s.logger.Info("Fetching dashboard statistics report", map[string]interface{}{ @@ -174,9 +191,37 @@ func (s *service) Statistics(ctx context.Context, query StatisticsQuery) (*Stati return nil, fmt.Errorf("dashboard statistics: %w", err) } + s.storeStatistics(days, out) return out, nil } +func (s *service) cachedStatistics(days int) (*StatisticsReportResponse, bool) { + now := time.Now() + + s.statisticsMu.Lock() + defer s.statisticsMu.Unlock() + + entry, ok := s.statistics[days] + if !ok || now.After(entry.expiresAt) { + if ok { + delete(s.statistics, days) + } + return nil, false + } + + return entry.value, true +} + +func (s *service) storeStatistics(days int, value *StatisticsReportResponse) { + s.statisticsMu.Lock() + defer s.statisticsMu.Unlock() + + s.statistics[days] = statisticsCacheEntry{ + expiresAt: time.Now().Add(statisticsCacheTTL), + value: value, + } +} + 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 f1e0ca6..e149d29 100644 --- a/internal/dashboard/statistics_repository.go +++ b/internal/dashboard/statistics_repository.go @@ -10,6 +10,7 @@ import ( "go.mongodb.org/mongo-driver/v2/bson" mongodriver "go.mongodb.org/mongo-driver/v2/mongo" + "golang.org/x/sync/errgroup" ) const noticesCollectionName = "notices" @@ -19,32 +20,66 @@ 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)) - scrapedTED, err := r.dailyCountByTimestamp(ctx, noticesCollectionName, bson.M{ - "source": notice.TenderSourceTEDScraper, - "processing_metadata.scraped_at": bson.M{"$gte": startUnix, "$gt": 0}, - }, "processing_metadata.scraped_at") - if err != nil { - return nil, fmt.Errorf("scraped ted per day: %w", err) - } + var ( + scrapedTED map[string]int64 + scrapedDocuments map[string]int64 + translatedNotices map[string]int64 + totalDocuments int64 + totalTranslated int64 + ) - scrapedDocuments, err := r.scrapedDocumentsPerDay(ctx, startUnix) - if err != nil { - return nil, fmt.Errorf("scraped documents per day: %w", err) - } + g, gctx := errgroup.WithContext(ctx) - translatedNotices, err := r.translatedNoticesPerDay(ctx, startDay, endDay) - if err != nil { - return nil, fmt.Errorf("translated notices per day: %w", err) - } + g.Go(func() error { + counts, err := r.dailyCountByTimestamp(gctx, noticesCollectionName, bson.M{ + "source": notice.TenderSourceTEDScraper, + "processing_metadata.scraped_at": bson.M{"$gte": startUnix, "$gt": 0}, + }, "processing_metadata.scraped_at") + if err != nil { + return fmt.Errorf("scraped ted per day: %w", err) + } + scrapedTED = counts + return nil + }) - totalDocuments, err := r.totalScrapedDocuments(ctx) - if err != nil { - return nil, fmt.Errorf("total scraped documents: %w", err) - } + g.Go(func() error { + counts, err := r.scrapedDocumentsPerDay(gctx, startUnix) + if err != nil { + return fmt.Errorf("scraped documents per day: %w", err) + } + scrapedDocuments = counts + return nil + }) - totalTranslated, err := r.metricsCounter.Get(ctx, orm.AITranslationSuccessCounterKey) - if err != nil { - return nil, fmt.Errorf("total translated tenders: %w", err) + g.Go(func() error { + counts, err := r.translatedNoticesPerDay(gctx, startDay, endDay) + if err != nil { + return fmt.Errorf("translated notices per day: %w", err) + } + translatedNotices = counts + return nil + }) + + g.Go(func() error { + total, err := r.totalScrapedDocuments(gctx) + if err != nil { + return fmt.Errorf("total scraped documents: %w", err) + } + totalDocuments = total + return nil + }) + + g.Go(func() error { + total, err := r.metricsCounter.Get(gctx, orm.AITranslationSuccessCounterKey) + if err != nil { + return fmt.Errorf("total translated tenders: %w", err) + } + totalTranslated = total + return nil + }) + + if err := g.Wait(); err != nil { + return nil, err } return &StatisticsReportResponse{ @@ -94,6 +129,10 @@ func (r *repository) scrapedDocumentsPerDay(ctx context.Context, startUnix int64 {{Key: "$match", Value: bson.M{ "processing_metadata.documents_scraped": true, "scraped_documents": bson.M{"$exists": true, "$ne": bson.A{}}, + "$or": []bson.M{ + {"processing_metadata.documents_scraped_at": bson.M{"$gte": startUnix}}, + {"scraped_documents.scraped_at": bson.M{"$gte": startUnix}}, + }, }}}, {{Key: "$unwind", Value: "$scraped_documents"}}, {{Key: "$addFields", Value: bson.M{ diff --git a/internal/notice/repository.go b/internal/notice/repository.go index 2d10bcf..2b6895d 100644 --- a/internal/notice/repository.go +++ b/internal/notice/repository.go @@ -39,6 +39,10 @@ func NewRepository(mongoManager *orm.ConnectionManager, logger logger.Logger) Re noticeIndexes := []orm.Index{ // processing_metadata.processed *orm.NewIndex("processing_metadata_processed_idx", bson.D{{Key: "processing_metadata.processed", Value: 1}}), + *orm.NewIndex("source_scraped_at_idx", bson.D{ + {Key: "source", Value: 1}, + {Key: "processing_metadata.scraped_at", Value: 1}, + }), *orm.NewIndex("main_classification_idx", bson.D{{Key: "main_classification", Value: 1}}), *orm.NewIndex("contract_notice_id_idx", bson.D{{Key: "contract_notice_id", Value: 1}}), // One row per TED ContractNoticeID when present (parallel scrape races). diff --git a/internal/tender/repository.go b/internal/tender/repository.go index fcd22d3..abe4559 100644 --- a/internal/tender/repository.go +++ b/internal/tender/repository.go @@ -150,6 +150,13 @@ func NewRepository(mongoManager *orm.ConnectionManager, logger logger.Logger) Te {Key: "tender_deadline", Value: 1}, {Key: "created_at", Value: 1}, }), + *orm.NewIndex("documents_scraped_at_idx", bson.D{ + {Key: "processing_metadata.documents_scraped", Value: 1}, + {Key: "processing_metadata.documents_scraped_at", Value: -1}, + }), + *orm.NewIndex("scraped_documents_scraped_at_idx", bson.D{ + {Key: "scraped_documents.scraped_at", Value: 1}, + }), // Index for CPV matching queries *orm.NewIndex("cpv_matching_idx", bson.D{ {Key: "status", Value: 1}, diff --git a/pkg/ai_summarizer/config.go b/pkg/ai_summarizer/config.go index 6b42e2d..c5141bb 100644 --- a/pkg/ai_summarizer/config.go +++ b/pkg/ai_summarizer/config.go @@ -26,7 +26,7 @@ type Config struct { MinioSecretKey string // MinIO secret access key MinioUseSSL bool // Whether to use SSL for MinIO connection MinioRegion string // MinIO region - MinioBucket string // MinIO bucket name (default: "opplens-documents") + MinioBucket string // MinIO bucket name (default: "opplens") MinioTimeout time.Duration // Timeout for MinIO operations } @@ -37,7 +37,7 @@ func DefaultConfig() *Config { APIRetryCount: 2, APIRetryDelay: 3 * time.Second, - MinioBucket: "opplens-documents", + MinioBucket: "opplens", MinioRegion: "us-east-1", MinioUseSSL: false, MinioTimeout: 30 * time.Second, diff --git a/pkg/ai_summarizer/storage.go b/pkg/ai_summarizer/storage.go index 2dd75e4..d51d314 100644 --- a/pkg/ai_summarizer/storage.go +++ b/pkg/ai_summarizer/storage.go @@ -19,7 +19,7 @@ import ( ) // procedurePrefix is prepended to every contract folder id in the MinIO key. -// The AI team's storage layout in bucket opplens-documents is: +// The AI team's storage layout in bucket opplens is: // // PROC_//tender.json // PROC_//documents/{files} diff --git a/pkg/mongo/counter.go b/pkg/mongo/counter.go index 3a5dc69..9d92e23 100644 --- a/pkg/mongo/counter.go +++ b/pkg/mongo/counter.go @@ -104,14 +104,47 @@ func (c *Counter) GetDailyCounts(ctx context.Context, startDay, endDay time.Time return map[string]int64{}, nil } - counts := make(map[string]int64) + keys := make([]string, 0) + dates := make([]string, 0) for day := startDay; !day.After(endDay); day = day.AddDate(0, 0, 1) { - date := day.Format("2006-01-02") - value, err := c.Get(ctx, AITranslationSuccessDailyCounterKey(day)) - if err != nil { - return nil, err + dates = append(dates, day.Format("2006-01-02")) + keys = append(keys, AITranslationSuccessDailyCounterKey(day)) + } + + counts := make(map[string]int64, len(dates)) + for _, date := range dates { + counts[date] = 0 + } + + cursor, err := c.mongo.GetCollection(metricsCountersCollection).Find(ctx, bson.M{ + "_id": bson.M{"$in": keys}, + }) + if err != nil { + return nil, fmt.Errorf("mongo counter get daily counts: %w", err) + } + defer cursor.Close(ctx) + + keyToDate := make(map[string]string, len(keys)) + for i, key := range keys { + keyToDate[key] = dates[i] + } + + for cursor.Next(ctx) { + var doc struct { + ID string `bson:"_id"` + Count int64 `bson:"count"` } - counts[date] = value + if err := cursor.Decode(&doc); err != nil { + return nil, fmt.Errorf("mongo counter decode daily count: %w", err) + } + date, ok := keyToDate[doc.ID] + if !ok { + continue + } + counts[date] = doc.Count + } + if err := cursor.Err(); err != nil { + return nil, fmt.Errorf("mongo counter iterate daily counts: %w", err) } return counts, nil