diff --git a/cmd/web/main.go b/cmd/web/main.go index a69678b..a81ab52 100644 --- a/cmd/web/main.go +++ b/cmd/web/main.go @@ -171,6 +171,7 @@ func main() { // Initialize AI Summarizer service var aiSummarizerClient tender.AISummarizerClient var aiSummarizerStorage tender.AISummarizerStorage + var procedureDocumentsLister dashboard.ProcedureDocumentsLister var aiRecommendationClient company.AIRecommendationClient var aiPipelineClient ai_pipeline.Client @@ -181,6 +182,7 @@ func main() { } if s := bootstrap.InitAISummarizerStorage(conf.AISummarizer, logger); s != nil { aiSummarizerStorage = s + procedureDocumentsLister = s } // Initialize authorization service @@ -230,7 +232,7 @@ func main() { cmsService := cms.NewService(cmsRepository, logger) kanbanService := kanban.NewService(boardRepository, columnRepository, cardRepository, goRulesClient, conf.GoRules.RuleID, logger) documentScraperService := document_scraper.NewService(tenderRepository, logger) - dashboardRepository := dashboard.NewRepository(mongoManager, logger) + dashboardRepository := dashboard.NewRepository(mongoManager, logger, procedureDocumentsLister) dashboardService := dashboard.NewService(dashboardRepository, logger) aiPipelineService := ai_pipeline.NewService(aiPipelineClient, logger, tenderService) logger.Info("Services initialized successfully", map[string]interface{}{ diff --git a/internal/dashboard/repository.go b/internal/dashboard/repository.go index 7587af3..d019b53 100644 --- a/internal/dashboard/repository.go +++ b/internal/dashboard/repository.go @@ -6,6 +6,7 @@ import ( "strings" "time" "tm/internal/tender" + "tm/pkg/ai_summarizer" "tm/pkg/logger" orm "tm/pkg/mongo" @@ -13,6 +14,12 @@ import ( "go.mongodb.org/mongo-driver/v2/mongo" ) +// ProcedureDocumentsLister lists contract folders that have scraped documents in MinIO. +// This is the source of truth for document-scrape statistics, matching tender panel search. +type ProcedureDocumentsLister interface { + ListProceduresWithDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error) +} + const defaultValueCurrency = "EUR" // Repository defines dashboard data access. @@ -27,19 +34,21 @@ type Repository interface { } type repository struct { - ormRepo orm.Repository[tender.Tender] - mongoManager *orm.ConnectionManager - metricsCounter *orm.Counter - logger logger.Logger + ormRepo orm.Repository[tender.Tender] + mongoManager *orm.ConnectionManager + metricsCounter *orm.Counter + procedureLister ProcedureDocumentsLister + logger logger.Logger } // NewRepository creates a dashboard repository backed by the tenders collection. -func NewRepository(mongoManager *orm.ConnectionManager, log logger.Logger) Repository { +func NewRepository(mongoManager *orm.ConnectionManager, log logger.Logger, procedureLister ProcedureDocumentsLister) Repository { return &repository{ - ormRepo: orm.NewRepository[tender.Tender](mongoManager.GetCollection(tenderCollectionName()), log), - mongoManager: mongoManager, - metricsCounter: orm.NewCounter(mongoManager), - logger: log, + ormRepo: orm.NewRepository[tender.Tender](mongoManager.GetCollection(tenderCollectionName()), log), + mongoManager: mongoManager, + metricsCounter: orm.NewCounter(mongoManager), + procedureLister: procedureLister, + logger: log, } } @@ -313,16 +322,16 @@ func (r *repository) ClosingSoon(ctx context.Context, limit int, windowSec int64 {{Key: "$sort", Value: bson.M{"effective_deadline": 1}}}, {{Key: "$limit", Value: limit}}, {{Key: "$project", Value: bson.M{ - "_id": 1, - "title": 1, - "country_code": 1, - "buyer_organization": 1, - "submission_deadline": 1, - "tender_deadline": 1, - "status": 1, - "estimated_value": 1, - "currency": 1, - "effective_deadline": 1, + "_id": 1, + "title": 1, + "country_code": 1, + "buyer_organization": 1, + "submission_deadline": 1, + "tender_deadline": 1, + "status": 1, + "estimated_value": 1, + "currency": 1, + "effective_deadline": 1, }}}, } @@ -351,12 +360,12 @@ func (r *repository) Recent(ctx context.Context, limit int, cursor string) ([]Re filter, orm.ListPaginationOptions{ Projection: bson.M{ - "title": 1, - "country_code": 1, - "status": 1, + "title": 1, + "country_code": 1, + "status": 1, "buyer_organization": 1, - "created_at": 1, - "publication_date": 1, + "created_at": 1, + "publication_date": 1, }, }, ) diff --git a/internal/dashboard/statistics_repository.go b/internal/dashboard/statistics_repository.go index ae7fe3d..a63cb42 100644 --- a/internal/dashboard/statistics_repository.go +++ b/internal/dashboard/statistics_repository.go @@ -3,6 +3,7 @@ package dashboard import ( "context" "fmt" + "strings" "time" "tm/internal/notice" @@ -15,11 +16,23 @@ import ( const noticesCollectionName = "notices" +// 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 { + match bson.M + zero bool +} + func (r *repository) Statistics(ctx context.Context, days int, startUnix int64) (*StatisticsReportResponse, error) { now := time.Now().UTC() 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) + if err != nil { + return nil, err + } + var ( scrapedTED map[string]int64 scrapedDocuments map[string]int64 @@ -45,7 +58,7 @@ func (r *repository) Statistics(ctx context.Context, days int, startUnix int64) }) g.Go(func() error { - counts, err := r.scrapedDocumentsPerDay(gctx, startUnix) + counts, err := r.scrapedDocumentsPerDay(gctx, startUnix, scrapedScope) if err != nil { return fmt.Errorf("scraped documents per day: %w", err) } @@ -63,9 +76,9 @@ func (r *repository) Statistics(ctx context.Context, days int, startUnix int64) }) g.Go(func() error { - total, err := r.totalScrapedDocuments(gctx) + total, err := r.totalScrapedTenders(gctx, scrapedScope) if err != nil { - return fmt.Errorf("total scraped documents: %w", err) + return fmt.Errorf("total scraped tenders: %w", err) } totalDocuments = total return nil @@ -126,24 +139,17 @@ func (r *repository) dailyCountByTimestamp(ctx context.Context, collection strin return decodeDailyCounts(ctx, cursor) } -func (r *repository) scrapedDocumentsPerDay(ctx context.Context, startUnix int64) (map[string]int64, error) { +func (r *repository) scrapedDocumentsPerDay(ctx context.Context, startUnix int64, scope scrapedTendersScope) (map[string]int64, error) { + if scope.zero { + return map[string]int64{}, nil + } + + // Daily buckets use documents_scraped_at only. Tenders present in MinIO but not yet + // synced to Mongo are included in the lifetime total but omitted from the daily series. pipeline := mongodriver.Pipeline{ - {{Key: "$match", Value: bson.M{ - "processing_metadata.documents_scraped": true, - "scraped_documents": bson.M{"$exists": true, "$ne": bson.A{}}, - }}}, - {{Key: "$unwind", Value: "$scraped_documents"}}, + {{Key: "$match", Value: scope.match}}, {{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"), + "metric_ts": normalizeTimestampExpr("processing_metadata.documents_scraped_at"), }}}, {{Key: "$match", Value: bson.M{ "metric_ts": bson.M{"$gte": startUnix, "$gt": 0}, @@ -173,37 +179,63 @@ func (r *repository) translatedNoticesPerDay(ctx context.Context, startDay, endD return r.metricsCounter.GetDailyCounts(ctx, startDay, endDay) } -func (r *repository) totalScrapedDocuments(ctx context.Context) (int64, error) { - pipeline := mongodriver.Pipeline{ - {{Key: "$match", Value: bson.M{ - "processing_metadata.documents_scraped": true, - "scraped_documents": bson.M{"$exists": true, "$ne": bson.A{}}, - }}}, - {{Key: "$project", Value: bson.M{ - "doc_count": bson.M{"$size": "$scraped_documents"}, - }}}, - {{Key: "$group", Value: bson.M{ - "_id": nil, - "total": bson.M{"$sum": "$doc_count"}, - }}}, - } - - cursor, err := r.mongoManager.GetCollection(tenderCollectionName()).Aggregate(ctx, pipeline) - if err != nil { - return 0, err - } - defer cursor.Close(ctx) - - if !cursor.Next(ctx) { +func (r *repository) totalScrapedTenders(ctx context.Context, scope scrapedTendersScope) (int64, error) { + if scope.zero { return 0, nil } - var row bson.M - if err := cursor.Decode(&row); err != nil { - return 0, err + return r.ormRepo.Count(ctx, scope.match) +} + +func (r *repository) resolveScrapedTendersScope(ctx context.Context) (scrapedTendersScope, error) { + folderIDs, err := r.scrapedDocumentsFolderIDs(ctx) + if err != nil { + return scrapedTendersScope{}, err + } + if len(folderIDs) > 0 { + return scrapedTendersScope{ + match: bson.M{"contract_folder_id": bson.M{"$in": folderIDs}}, + }, nil + } + if r.procedureLister != nil { + return scrapedTendersScope{zero: true}, nil + } + return scrapedTendersScope{ + match: bson.M{ + "processing_metadata.documents_scraped": true, + "scraped_documents": bson.M{"$exists": true, "$ne": bson.A{}}, + }, + }, nil +} + +func (r *repository) scrapedDocumentsFolderIDs(ctx context.Context) ([]string, error) { + if r.procedureLister == nil { + return nil, nil } - return aggregateCount(row, "total"), nil + procedures, err := r.procedureLister.ListProceduresWithDocuments(ctx) + if err != nil { + return nil, err + } + + folderIDs := make([]string, 0, len(procedures)) + seen := make(map[string]struct{}, len(procedures)) + for _, proc := range procedures { + if proc.DocumentCount <= 0 { + continue + } + id := strings.TrimSpace(proc.ContractFolderID) + if id == "" { + continue + } + if _, ok := seen[id]; ok { + continue + } + seen[id] = struct{}{} + folderIDs = append(folderIDs, id) + } + + return folderIDs, nil } func decodeDailyCounts(ctx context.Context, cursor *mongodriver.Cursor) (map[string]int64, error) { diff --git a/internal/dashboard/statistics_repository_test.go b/internal/dashboard/statistics_repository_test.go new file mode 100644 index 0000000..ac9302f --- /dev/null +++ b/internal/dashboard/statistics_repository_test.go @@ -0,0 +1,131 @@ +package dashboard + +import ( + "context" + "errors" + "testing" + + "tm/pkg/ai_summarizer" + + "go.mongodb.org/mongo-driver/v2/bson" +) + +type mockProcedureDocumentsLister struct { + procedures []ai_summarizer.ProcedureDocumentsSummary + err error +} + +func (m *mockProcedureDocumentsLister) ListProceduresWithDocuments(context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error) { + if m.err != nil { + return nil, m.err + } + return m.procedures, nil +} + +func TestScrapedDocumentsFolderIDsUsesMinIOProcedures(t *testing.T) { + repo := &repository{ + procedureLister: &mockProcedureDocumentsLister{ + procedures: []ai_summarizer.ProcedureDocumentsSummary{ + {ContractFolderID: "PROC-1", DocumentCount: 3}, + {ContractFolderID: "PROC-2", DocumentCount: 0}, + {ContractFolderID: " PROC-3 ", DocumentCount: 1}, + {ContractFolderID: "PROC-1", DocumentCount: 2}, + }, + }, + } + + got, err := repo.scrapedDocumentsFolderIDs(context.Background()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + want := []string{"PROC-1", "PROC-3"} + if len(got) != len(want) { + t.Fatalf("expected %d folder IDs, got %d: %v", len(want), len(got), got) + } + for i, id := range want { + if got[i] != id { + t.Fatalf("expected folder ID %q at index %d, got %q", id, i, got[i]) + } + } +} + +func TestScrapedDocumentsFolderIDsWithoutLister(t *testing.T) { + repo := &repository{} + + got, err := repo.scrapedDocumentsFolderIDs(context.Background()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got != nil { + t.Fatalf("expected nil folder IDs without lister, got %v", got) + } +} + +func TestScrapedDocumentsFolderIDsPropagatesError(t *testing.T) { + repo := &repository{ + procedureLister: &mockProcedureDocumentsLister{ + err: errors.New("minio unavailable"), + }, + } + + _, err := repo.scrapedDocumentsFolderIDs(context.Background()) + if err == nil { + t.Fatal("expected error") + } +} + +func TestResolveScrapedTendersScopeUsesMinIOFolders(t *testing.T) { + repo := &repository{ + procedureLister: &mockProcedureDocumentsLister{ + procedures: []ai_summarizer.ProcedureDocumentsSummary{ + {ContractFolderID: "PROC-1", DocumentCount: 2}, + }, + }, + } + + scope, err := repo.resolveScrapedTendersScope(context.Background()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if scope.zero { + t.Fatal("expected non-zero scope") + } + in, ok := scope.match["contract_folder_id"].(bson.M) + if !ok { + t.Fatalf("expected contract_folder_id filter, got %#v", scope.match) + } + ids, ok := in["$in"].([]string) + if !ok || len(ids) != 1 || ids[0] != "PROC-1" { + t.Fatalf("unexpected folder IDs: %#v", in["$in"]) + } +} + +func TestResolveScrapedTendersScopeZeroWhenMinIOEmpty(t *testing.T) { + repo := &repository{ + procedureLister: &mockProcedureDocumentsLister{}, + } + + scope, err := repo.resolveScrapedTendersScope(context.Background()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !scope.zero { + t.Fatalf("expected zero scope, got %#v", scope) + } +} + +func TestResolveScrapedTendersScopeMongoFallbackWithoutLister(t *testing.T) { + repo := &repository{} + + scope, err := repo.resolveScrapedTendersScope(context.Background()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if scope.zero { + t.Fatal("expected mongo fallback scope") + } + if scope.match["processing_metadata.documents_scraped"] != true { + t.Fatalf("expected mongo fallback filter, got %#v", scope.match) + } +}