diff --git a/internal/dashboard/repository.go b/internal/dashboard/repository.go index bc7171b..7542048 100644 --- a/internal/dashboard/repository.go +++ b/internal/dashboard/repository.go @@ -22,6 +22,11 @@ type ProcedureDocumentsLister interface { ListProceduresWithDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error) } +// scrapedDocumentsScanner optionally scans MinIO for per-day document counts in the same pass. +type scrapedDocumentsScanner interface { + ScanScrapedDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, map[string]int64, error) +} + const defaultValueCurrency = "EUR" // Repository defines dashboard data access. diff --git a/internal/dashboard/statistics_repository.go b/internal/dashboard/statistics_repository.go index 764549d..b5320bf 100644 --- a/internal/dashboard/statistics_repository.go +++ b/internal/dashboard/statistics_repository.go @@ -7,6 +7,7 @@ import ( "time" "tm/internal/notice" + "tm/pkg/ai_summarizer" orm "tm/pkg/mongo" "go.mongodb.org/mongo-driver/v2/bson" @@ -29,8 +30,10 @@ const ( // 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 + match bson.M + zero bool + fromMinIO bool + dailyCounts map[string]int64 // non-nil when scope was resolved from MinIO } func (r *repository) Statistics(ctx context.Context, days int, startUnix int64) (*StatisticsReportResponse, error) { @@ -158,12 +161,42 @@ func (r *repository) scrapedDocumentsPerDay(ctx context.Context, startUnix int64 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. + if scope.dailyCounts != nil { + return filterDailyCountsSince(scope.dailyCounts, startUnix), nil + } + + return r.scrapedDocumentsPerDayFromMongo(ctx, startUnix, scope.match) +} + +func filterDailyCountsSince(counts map[string]int64, startUnix int64) map[string]int64 { + if len(counts) == 0 { + return map[string]int64{} + } + startDay := time.Unix(startUnix, 0).UTC().Format("2006-01-02") + filtered := make(map[string]int64) + for date, count := range counts { + if date >= startDay { + filtered[date] = count + } + } + return filtered +} + +func (r *repository) scrapedDocumentsPerDayFromMongo(ctx context.Context, startUnix int64, match bson.M) (map[string]int64, error) { pipeline := mongodriver.Pipeline{ - {{Key: "$match", Value: scope.match}}, + {{Key: "$match", Value: match}}, + {{Key: "$unwind", Value: "$scraped_documents"}}, {{Key: "$addFields", Value: bson.M{ - "metric_ts": normalizeTimestampExpr("processing_metadata.documents_scraped_at"), + "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}, @@ -257,35 +290,46 @@ func (r *repository) cachedScrapedTendersScope(ctx context.Context) (scrapedTend } func (r *repository) resolveScrapedTendersScope(ctx context.Context) (scrapedTendersScope, error) { - folderIDs, err := r.scrapedDocumentsFolderIDs(ctx) + procedures, dailyCounts, err := r.listScrapedProcedures(ctx) if err != nil { return scrapedTendersScope{}, err } + + folderIDs := folderIDsFromProcedures(procedures) if len(folderIDs) > 0 { if len(folderIDs) > maxScrapedFolderInClause { r.logger.Warn("Too many MinIO procedure folders for dashboard statistics, using Mongo fallback", map[string]interface{}{ - "folder_count": len(folderIDs), + "folder_count": len(folderIDs), "max_in_clause": maxScrapedFolderInClause, }) return mongoScrapedTendersScope(), nil } return scrapedTendersScope{ - match: bson.M{"contract_folder_id": bson.M{"$in": folderIDs}}, + match: bson.M{"contract_folder_id": bson.M{"$in": folderIDs}}, + fromMinIO: true, + dailyCounts: dailyCounts, }, nil } return mongoScrapedTendersScope(), nil } -func (r *repository) scrapedDocumentsFolderIDs(ctx context.Context) ([]string, error) { +func (r *repository) listScrapedProcedures(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, map[string]int64, error) { if r.procedureLister == nil { - return nil, nil + return nil, nil, nil + } + + if scanner, ok := r.procedureLister.(scrapedDocumentsScanner); ok { + return scanner.ScanScrapedDocuments(ctx) } procedures, err := r.procedureLister.ListProceduresWithDocuments(ctx) if err != nil { - return nil, err + return nil, nil, err } + return procedures, nil, nil +} +func folderIDsFromProcedures(procedures []ai_summarizer.ProcedureDocumentsSummary) []string { folderIDs := make([]string, 0, len(procedures)) seen := make(map[string]struct{}, len(procedures)) for _, proc := range procedures { @@ -302,8 +346,19 @@ func (r *repository) scrapedDocumentsFolderIDs(ctx context.Context) ([]string, e seen[id] = struct{}{} folderIDs = append(folderIDs, id) } + return folderIDs +} - return folderIDs, nil +func (r *repository) scrapedDocumentsFolderIDs(ctx context.Context) ([]string, error) { + if r.procedureLister == nil { + return nil, nil + } + + procedures, _, err := r.listScrapedProcedures(ctx) + if err != nil { + return nil, err + } + return folderIDsFromProcedures(procedures), 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 index f76ea3b..77eef99 100644 --- a/internal/dashboard/statistics_repository_test.go +++ b/internal/dashboard/statistics_repository_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "testing" + "time" "tm/pkg/ai_summarizer" @@ -11,8 +12,9 @@ import ( ) type mockProcedureDocumentsLister struct { - procedures []ai_summarizer.ProcedureDocumentsSummary - err error + procedures []ai_summarizer.ProcedureDocumentsSummary + dailyCounts map[string]int64 + err error } func (m *mockProcedureDocumentsLister) ListProceduresWithDocuments(context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error) { @@ -22,6 +24,13 @@ func (m *mockProcedureDocumentsLister) ListProceduresWithDocuments(context.Conte return m.procedures, nil } +func (m *mockProcedureDocumentsLister) ScanScrapedDocuments(context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, map[string]int64, error) { + if m.err != nil { + return nil, nil, m.err + } + return m.procedures, m.dailyCounts, nil +} + func TestScrapedDocumentsFolderIDsUsesMinIOProcedures(t *testing.T) { repo := &repository{ procedureLister: &mockProcedureDocumentsLister{ @@ -91,6 +100,9 @@ func TestResolveScrapedTendersScopeUsesMinIOFolders(t *testing.T) { if scope.zero { t.Fatal("expected non-zero scope") } + if !scope.fromMinIO { + t.Fatal("expected MinIO-backed scope") + } in, ok := scope.match["contract_folder_id"].(bson.M) if !ok { t.Fatalf("expected contract_folder_id filter, got %#v", scope.match) @@ -133,6 +145,52 @@ func TestResolveScrapedTendersScopeMongoFallbackWithoutLister(t *testing.T) { } } +func TestFilterDailyCountsSince(t *testing.T) { + counts := map[string]int64{ + "2026-06-01": 2, + "2026-06-15": 5, + "2026-06-28": 1, + } + + startUnix := time.Date(2026, 6, 15, 0, 0, 0, 0, time.UTC).Unix() + got := filterDailyCountsSince(counts, startUnix) + + want := map[string]int64{ + "2026-06-15": 5, + "2026-06-28": 1, + } + if len(got) != len(want) { + t.Fatalf("expected %d dates, got %d: %v", len(want), len(got), got) + } + for date, count := range want { + if got[date] != count { + t.Fatalf("expected %d on %s, got %d", count, date, got[date]) + } + } +} + +func TestScrapedDocumentsPerDayUsesMinIODailyCounts(t *testing.T) { + repo := &repository{} + scope := scrapedTendersScope{ + dailyCounts: map[string]int64{ + "2026-06-01": 2, + "2026-06-28": 3, + }, + } + startUnix := time.Date(2026, 6, 15, 0, 0, 0, 0, time.UTC).Unix() + + got, err := repo.scrapedDocumentsPerDay(context.Background(), startUnix, scope) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got["2026-06-01"] != 0 { + t.Fatalf("expected old date filtered out, got %v", got) + } + if got["2026-06-28"] != 3 { + t.Fatalf("expected 3 on 2026-06-28, got %v", got) + } +} + func TestMongoScrapedTendersScopeMatchesStatisticsFilter(t *testing.T) { scope := mongoScrapedTendersScope() if scope.zero { diff --git a/pkg/ai_summarizer/storage.go b/pkg/ai_summarizer/storage.go index d51d314..eeb6531 100644 --- a/pkg/ai_summarizer/storage.go +++ b/pkg/ai_summarizer/storage.go @@ -570,12 +570,20 @@ func (s *StorageClient) ListDocuments(ctx context.Context, contractFolderID, not // ListProceduresWithDocuments scans MinIO for PROC_/…/documents/* keys // and returns one entry per contract folder id that has at least one document file. func (s *StorageClient) ListProceduresWithDocuments(ctx context.Context) ([]ProcedureDocumentsSummary, error) { - s.logger.Debug("Listing procedure folders with documents from MinIO", map[string]interface{}{ + procedures, _, err := s.ScanScrapedDocuments(ctx) + return procedures, err +} + +// ScanScrapedDocuments scans MinIO once and returns per-procedure summaries plus +// daily document counts keyed by UTC date (YYYY-MM-DD). +func (s *StorageClient) ScanScrapedDocuments(ctx context.Context) ([]ProcedureDocumentsSummary, map[string]int64, error) { + s.logger.Debug("Scanning scraped documents from MinIO", map[string]interface{}{ "bucket": s.config.MinioBucket, "prefix": procedurePrefix, }) byFolder := make(map[string]*ProcedureDocumentsSummary) + dailyCounts := make(map[string]int64) var fatalErr error appendFromPrefix := func(prefix string) bool { @@ -603,24 +611,28 @@ func (s *StorageClient) ListProceduresWithDocuments(ctx context.Context) ([]Proc byFolder[contractFolderID] = entry } entry.DocumentCount++ - modified := object.LastModified.Unix() + modified := normalizeUnixSeconds(object.LastModified.Unix()) if modified > entry.LatestModified { entry.LatestModified = modified } + if modified > 0 { + date := time.Unix(modified, 0).UTC().Format("2006-01-02") + dailyCounts[date]++ + } } return true } if !appendFromPrefix(procedurePrefix) { - return nil, fatalErr + return nil, nil, fatalErr } if len(byFolder) == 0 { if !appendFromPrefix("") { - return nil, fatalErr + return nil, nil, fatalErr } } if fatalErr != nil { - return nil, fatalErr + return nil, nil, fatalErr } results := make([]ProcedureDocumentsSummary, 0, len(byFolder)) @@ -634,12 +646,19 @@ func (s *StorageClient) ListProceduresWithDocuments(ctx context.Context) ([]Proc return results[i].LatestModified > results[j].LatestModified }) - s.logger.Info("Listed procedure folders with documents from storage", map[string]interface{}{ - "bucket": s.config.MinioBucket, + s.logger.Info("Scanned scraped documents from storage", map[string]interface{}{ + "bucket": s.config.MinioBucket, "procedure_count": len(results), }) - return results, nil + return results, dailyCounts, nil +} + +func normalizeUnixSeconds(ts int64) int64 { + if ts > 1_000_000_000_000 { + return ts / 1000 + } + return ts } // contractFolderIDFromDocumentKey parses PROC_//documents/.