From 9676f9930417b9a62808c5cd2d34301ed36bf85e Mon Sep 17 00:00:00 2001 From: Mazyar Date: Wed, 17 Jun 2026 14:17:41 +0330 Subject: [PATCH 1/2] Refactor dashboard repository to integrate ProcedureDocumentsLister - Introduced the ProcedureDocumentsLister interface to list contract folders with scraped documents, enhancing the accuracy of document-scrape statistics. - Updated the dashboard repository to accept ProcedureDocumentsLister as a dependency, allowing for improved data retrieval. - Implemented tests for the new functionality, ensuring proper handling of scraped document folder IDs and error propagation. This update enhances the dashboard's capability to manage and report on scraped documents, improving overall system efficiency and data integrity. --- cmd/web/main.go | 4 +- internal/dashboard/repository.go | 57 +++++----- internal/dashboard/statistics_repository.go | 102 +++++++++++------- .../dashboard/statistics_repository_test.go | 74 +++++++++++++ 4 files changed, 172 insertions(+), 65 deletions(-) create mode 100644 internal/dashboard/statistics_repository_test.go 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..9c1a34d 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" @@ -63,9 +64,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) if err != nil { - return fmt.Errorf("total scraped documents: %w", err) + return fmt.Errorf("total scraped tenders: %w", err) } totalDocuments = total return nil @@ -127,23 +128,18 @@ func (r *repository) dailyCountByTimestamp(ctx context.Context, collection strin } func (r *repository) scrapedDocumentsPerDay(ctx context.Context, startUnix int64) (map[string]int64, error) { + match, err := r.scrapedTendersMatchFilter(ctx) + if err != nil { + return nil, err + } + if len(match) == 0 { + return map[string]int64{}, nil + } + 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: 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 +169,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) +func (r *repository) totalScrapedTenders(ctx context.Context) (int64, error) { + match, err := r.scrapedTendersMatchFilter(ctx) if err != nil { return 0, err } - defer cursor.Close(ctx) - - if !cursor.Next(ctx) { + if len(match) == 0 { return 0, nil } - var row bson.M - if err := cursor.Decode(&row); err != nil { - return 0, err + return r.ormRepo.Count(ctx, match) +} + +func (r *repository) scrapedTendersMatchFilter(ctx context.Context) (bson.M, error) { + folderIDs, err := r.scrapedDocumentsFolderIDs(ctx) + if err != nil { + return nil, err + } + if len(folderIDs) > 0 { + return bson.M{"contract_folder_id": bson.M{"$in": folderIDs}}, nil + } + if r.procedureLister != nil { + return bson.M{}, nil + } + return 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..ed09c4e --- /dev/null +++ b/internal/dashboard/statistics_repository_test.go @@ -0,0 +1,74 @@ +package dashboard + +import ( + "context" + "errors" + "testing" + + "tm/pkg/ai_summarizer" +) + +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") + } +} From 8035118f44cec6e03e0779d8a8d3c131797718e4 Mon Sep 17 00:00:00 2001 From: Mazyar Date: Wed, 17 Jun 2026 15:01:52 +0330 Subject: [PATCH 2/2] Enhance dashboard statistics repository with scraped tenders scope resolution - Introduced a new `scrapedTendersScope` type to encapsulate the MongoDB filter for tenders with scraped documents, improving clarity and maintainability. - Updated the `Statistics` method to utilize the new scope resolution, allowing for more accurate data retrieval based on the presence of scraped documents. - Implemented multiple tests for the `resolveScrapedTendersScope` method, ensuring correct behavior for various scenarios, including empty and fallback cases. This update enhances the dashboard's ability to manage scraped document statistics, improving overall data accuracy and system performance. --- internal/dashboard/statistics_repository.go | 56 ++++++++++-------- .../dashboard/statistics_repository_test.go | 57 +++++++++++++++++++ 2 files changed, 90 insertions(+), 23 deletions(-) diff --git a/internal/dashboard/statistics_repository.go b/internal/dashboard/statistics_repository.go index 9c1a34d..a63cb42 100644 --- a/internal/dashboard/statistics_repository.go +++ b/internal/dashboard/statistics_repository.go @@ -16,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 @@ -46,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) } @@ -64,7 +76,7 @@ func (r *repository) Statistics(ctx context.Context, days int, startUnix int64) }) g.Go(func() error { - total, err := r.totalScrapedTenders(gctx) + total, err := r.totalScrapedTenders(gctx, scrapedScope) if err != nil { return fmt.Errorf("total scraped tenders: %w", err) } @@ -127,17 +139,15 @@ 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) { - match, err := r.scrapedTendersMatchFilter(ctx) - if err != nil { - return nil, err - } - if len(match) == 0 { +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: match}}, + {{Key: "$match", Value: scope.match}}, {{Key: "$addFields", Value: bson.M{ "metric_ts": normalizeTimestampExpr("processing_metadata.documents_scraped_at"), }}}, @@ -169,32 +179,32 @@ func (r *repository) translatedNoticesPerDay(ctx context.Context, startDay, endD return r.metricsCounter.GetDailyCounts(ctx, startDay, endDay) } -func (r *repository) totalScrapedTenders(ctx context.Context) (int64, error) { - match, err := r.scrapedTendersMatchFilter(ctx) - if err != nil { - return 0, err - } - if len(match) == 0 { +func (r *repository) totalScrapedTenders(ctx context.Context, scope scrapedTendersScope) (int64, error) { + if scope.zero { return 0, nil } - return r.ormRepo.Count(ctx, match) + return r.ormRepo.Count(ctx, scope.match) } -func (r *repository) scrapedTendersMatchFilter(ctx context.Context) (bson.M, error) { +func (r *repository) resolveScrapedTendersScope(ctx context.Context) (scrapedTendersScope, error) { folderIDs, err := r.scrapedDocumentsFolderIDs(ctx) if err != nil { - return nil, err + return scrapedTendersScope{}, err } if len(folderIDs) > 0 { - return bson.M{"contract_folder_id": bson.M{"$in": folderIDs}}, nil + return scrapedTendersScope{ + match: bson.M{"contract_folder_id": bson.M{"$in": folderIDs}}, + }, nil } if r.procedureLister != nil { - return bson.M{}, nil + return scrapedTendersScope{zero: true}, nil } - return bson.M{ - "processing_metadata.documents_scraped": true, - "scraped_documents": bson.M{"$exists": true, "$ne": bson.A{}}, + return scrapedTendersScope{ + match: bson.M{ + "processing_metadata.documents_scraped": true, + "scraped_documents": bson.M{"$exists": true, "$ne": bson.A{}}, + }, }, nil } diff --git a/internal/dashboard/statistics_repository_test.go b/internal/dashboard/statistics_repository_test.go index ed09c4e..ac9302f 100644 --- a/internal/dashboard/statistics_repository_test.go +++ b/internal/dashboard/statistics_repository_test.go @@ -6,6 +6,8 @@ import ( "testing" "tm/pkg/ai_summarizer" + + "go.mongodb.org/mongo-driver/v2/bson" ) type mockProcedureDocumentsLister struct { @@ -72,3 +74,58 @@ func TestScrapedDocumentsFolderIDsPropagatesError(t *testing.T) { 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) + } +}