diff --git a/internal/tender/documents_scraped_filter.go b/internal/tender/documents_scraped_filter.go new file mode 100644 index 0000000..8265fbc --- /dev/null +++ b/internal/tender/documents_scraped_filter.go @@ -0,0 +1,117 @@ +package tender + +import ( + "context" + "errors" + "fmt" + "strings" + "time" + + "tm/pkg/ai_summarizer" +) + +const ( + // contractFoldersWithDocsCacheTTL avoids scanning MinIO on every documents_scraped list request. + contractFoldersWithDocsCacheTTL = 60 * time.Second + // contractFoldersWithDocsResolveTimeout bounds a single MinIO procedure scan. + contractFoldersWithDocsResolveTimeout = 3 * time.Minute +) + +type contractFoldersWithDocsCacheEntry struct { + expiresAt time.Time + folderIDs []string + valid bool +} + +// enrichDocumentsScrapedFilter resolves contract_folder_id values that currently have +// documents in MinIO and stores them on the search form for the repository filter. +func (s *tenderService) enrichDocumentsScrapedFilter(ctx context.Context, form *SearchForm) error { + if form == nil || !form.DocumentsScraped { + return nil + } + + folderIDs, err := s.cachedContractFolderIDsWithDocuments(ctx) + if err != nil { + s.logger.Error("Failed to resolve tenders with scraped documents from MinIO", map[string]interface{}{ + "error": err.Error(), + }) + if errors.Is(err, ai_summarizer.ErrMinIOUnavailable) { + return fmt.Errorf("failed to resolve tenders with scraped documents: MinIO connection unavailable: %w", err) + } + return fmt.Errorf("failed to resolve tenders with scraped documents: %w", err) + } + + form.ContractFolderIDsWithDocuments = folderIDs + return nil +} + +func (s *tenderService) cachedContractFolderIDsWithDocuments(ctx context.Context) ([]string, error) { + now := time.Now() + + s.contractFoldersCacheMu.Lock() + if s.contractFoldersCache.valid && now.Before(s.contractFoldersCache.expiresAt) { + ids := append([]string(nil), s.contractFoldersCache.folderIDs...) + s.contractFoldersCacheMu.Unlock() + return ids, nil + } + s.contractFoldersCacheMu.Unlock() + + v, err, _ := s.contractFoldersCacheGroup.Do("contract-folders-with-docs", func() (interface{}, error) { + s.contractFoldersCacheMu.Lock() + if s.contractFoldersCache.valid && time.Now().Before(s.contractFoldersCache.expiresAt) { + ids := append([]string(nil), s.contractFoldersCache.folderIDs...) + s.contractFoldersCacheMu.Unlock() + return ids, nil + } + s.contractFoldersCacheMu.Unlock() + + resolveCtx, cancel := context.WithTimeout(ctx, contractFoldersWithDocsResolveTimeout) + defer cancel() + + procedures, listErr := s.listProceduresWithDocuments(resolveCtx) + if listErr != nil { + return nil, listErr + } + + folderIDs := contractFolderIDsFromProcedures(procedures) + + s.contractFoldersCacheMu.Lock() + s.contractFoldersCache = contractFoldersWithDocsCacheEntry{ + expiresAt: time.Now().Add(contractFoldersWithDocsCacheTTL), + folderIDs: folderIDs, + valid: true, + } + s.contractFoldersCacheMu.Unlock() + + return append([]string(nil), folderIDs...), nil + }) + if err != nil { + return nil, err + } + + ids, ok := v.([]string) + if !ok { + return nil, fmt.Errorf("unexpected contract folder cache value type %T", v) + } + return append([]string(nil), ids...), nil +} + +func contractFolderIDsFromProcedures(procedures []ai_summarizer.ProcedureDocumentsSummary) []string { + 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 +} diff --git a/internal/tender/documents_scraped_filter_test.go b/internal/tender/documents_scraped_filter_test.go new file mode 100644 index 0000000..33e8eaa --- /dev/null +++ b/internal/tender/documents_scraped_filter_test.go @@ -0,0 +1,181 @@ +package tender + +import ( + "context" + "errors" + "fmt" + "io" + "sync/atomic" + "testing" + "time" + + "go.mongodb.org/mongo-driver/v2/bson" + + "tm/pkg/ai_summarizer" +) + +type mockProcedureDocumentLister struct { + procedures []ai_summarizer.ProcedureDocumentsSummary + err error + calls atomic.Int32 +} + +func (m *mockProcedureDocumentLister) ListProceduresWithDocuments(context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error) { + m.calls.Add(1) + if m.err != nil { + return nil, m.err + } + return m.procedures, nil +} + +type stubProcedureDocumentStorage struct { + lister *mockProcedureDocumentLister +} + +func (s stubProcedureDocumentStorage) GetSummaryFromStorage(context.Context, string, string) (string, error) { + return "", ai_summarizer.ErrObjectNotFound +} + +func (s stubProcedureDocumentStorage) GetDocumentSummariesFromStorage(context.Context, string, string) ([]ai_summarizer.DocumentSummary, error) { + return nil, nil +} + +func (s stubProcedureDocumentStorage) GetTranslationFromStorage(context.Context, string, string, string) (ai_summarizer.StoredTranslation, error) { + return ai_summarizer.StoredTranslation{}, ai_summarizer.ErrTranslationNotReady +} + +func (s stubProcedureDocumentStorage) ListDocuments(context.Context, string, string) ([]ai_summarizer.StoredDocument, error) { + return nil, nil +} + +func (s stubProcedureDocumentStorage) DownloadDocument(context.Context, string) (io.ReadCloser, error) { + return nil, ai_summarizer.ErrObjectNotFound +} + +func (s stubProcedureDocumentStorage) ListProceduresWithDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error) { + return s.lister.ListProceduresWithDocuments(ctx) +} + +func TestBuildSearchFilterDocumentsScrapedUsesMinIOFolderIDs(t *testing.T) { + repo := &tenderRepository{} + filter := repo.buildSearchFilter(&SearchForm{ + DocumentsScraped: true, + ContractFolderIDsWithDocuments: []string{"folder-a", "folder-b"}, + }) + + in, ok := filter["contract_folder_id"].(bson.M)["$in"].([]string) + if !ok { + t.Fatalf("expected contract_folder_id $in filter, got %#v", filter) + } + if len(in) != 2 || in[0] != "folder-a" || in[1] != "folder-b" { + t.Fatalf("unexpected folder ids: %#v", in) + } + if filter["processing_metadata.documents_scraped"] != nil { + t.Fatalf("expected no stale mongo documents_scraped flag, got %#v", filter) + } +} + +func TestContractFolderIDsSearchClauseChunksLargeLists(t *testing.T) { + folderIDs := make([]string, maxContractFolderInClause+2) + for i := range folderIDs { + folderIDs[i] = fmt.Sprintf("folder-%d", i) + } + + clause := contractFolderIDsSearchClause(folderIDs) + or, ok := clause["$or"].([]bson.M) + if !ok || len(or) != 2 { + t.Fatalf("expected chunked $or filter, got %#v", clause) + } +} + +func TestCachedContractFolderIDsWithDocumentsUsesCache(t *testing.T) { + lister := &mockProcedureDocumentLister{ + procedures: []ai_summarizer.ProcedureDocumentsSummary{ + {ContractFolderID: "folder-1", DocumentCount: 2}, + {ContractFolderID: "folder-2", DocumentCount: 0}, + {ContractFolderID: "folder-3", DocumentCount: 1}, + }, + } + + svc := &tenderService{ + aiSummarizerStorage: stubProcedureDocumentStorage{lister: lister}, + } + + first, err := svc.cachedContractFolderIDsWithDocuments(context.Background()) + if err != nil { + t.Fatalf("first lookup failed: %v", err) + } + if len(first) != 2 || first[0] != "folder-1" || first[1] != "folder-3" { + t.Fatalf("unexpected folder ids: %#v", first) + } + if lister.calls.Load() != 1 { + t.Fatalf("expected one MinIO scan, got %d", lister.calls.Load()) + } + + second, err := svc.cachedContractFolderIDsWithDocuments(context.Background()) + if err != nil { + t.Fatalf("second lookup failed: %v", err) + } + if len(second) != 2 { + t.Fatalf("unexpected cached folder ids: %#v", second) + } + if lister.calls.Load() != 1 { + t.Fatalf("expected cache hit without extra MinIO scan, got %d calls", lister.calls.Load()) + } +} + +func TestEnrichDocumentsScrapedFilterPropagatesMinIOError(t *testing.T) { + svc := &tenderService{ + aiSummarizerStorage: stubProcedureDocumentStorage{ + lister: &mockProcedureDocumentLister{ + err: ai_summarizer.ErrMinIOUnavailable, + }, + }, + logger: noopTenderTestLogger{}, + } + + form := &SearchForm{DocumentsScraped: true} + err := svc.enrichDocumentsScrapedFilter(context.Background(), form) + if err == nil { + t.Fatal("expected MinIO error") + } + if !errors.Is(err, ai_summarizer.ErrMinIOUnavailable) { + t.Fatalf("expected wrapped MinIO unavailable error, got %v", err) + } +} + +func TestContractFolderIDsFromProceduresDedupesAndSkipsEmpty(t *testing.T) { + ids := contractFolderIDsFromProcedures([]ai_summarizer.ProcedureDocumentsSummary{ + {ContractFolderID: "a", DocumentCount: 1}, + {ContractFolderID: "a", DocumentCount: 2}, + {ContractFolderID: " ", DocumentCount: 1}, + {ContractFolderID: "b", DocumentCount: 0}, + }) + if len(ids) != 1 || ids[0] != "a" { + t.Fatalf("unexpected ids: %#v", ids) + } +} + +func TestContractFoldersWithDocsCacheExpires(t *testing.T) { + lister := &mockProcedureDocumentLister{ + procedures: []ai_summarizer.ProcedureDocumentsSummary{ + {ContractFolderID: "folder-1", DocumentCount: 1}, + }, + } + svc := &tenderService{aiSummarizerStorage: stubProcedureDocumentStorage{lister: lister}} + + if _, err := svc.cachedContractFolderIDsWithDocuments(context.Background()); err != nil { + t.Fatalf("lookup failed: %v", err) + } + + svc.contractFoldersCacheMu.Lock() + svc.contractFoldersCache.expiresAt = time.Now().Add(-time.Second) + svc.contractFoldersCacheMu.Unlock() + + if _, err := svc.cachedContractFolderIDsWithDocuments(context.Background()); err != nil { + t.Fatalf("lookup after expiry failed: %v", err) + } + if lister.calls.Load() != 2 { + t.Fatalf("expected cache refresh after expiry, got %d calls", lister.calls.Load()) + } +} diff --git a/internal/tender/form.go b/internal/tender/form.go index 5eea71a..4b62a3e 100644 --- a/internal/tender/form.go +++ b/internal/tender/form.go @@ -64,7 +64,7 @@ type SearchForm struct { OnlyActiveDeadlines bool `query:"only_active_deadlines" valid:"optional"` Language *string `query:"lang" valid:"optional"` DocumentsScraped bool `query:"documents_scraped" valid:"optional"` - // ContractFolderIDsWithDocuments is populated by the service from MinIO when DocumentsScraped is true. + // ContractFolderIDsWithDocuments is populated by the service from a cached MinIO scan when DocumentsScraped is true. ContractFolderIDsWithDocuments []string `query:"-" valid:"optional"` // ExcludeRejectedTenders hides company-rejected tenders from recommendation results. ExcludeRejectedTenders bool `query:"-" valid:"optional"` diff --git a/internal/tender/handler.go b/internal/tender/handler.go index e872be2..8dd5d45 100644 --- a/internal/tender/handler.go +++ b/internal/tender/handler.go @@ -160,7 +160,7 @@ func (h *TenderHandler) Delete(c echo.Context) error { // @Param languages query []string false "Filter by languages (comma-separated)" // @Param buyer_organization_id query string false "Filter by buyer organization ID" // @Param notice_publication_id query string false "Filter by TED notice publication ID" -// @Param documents_scraped query bool false "When true, only tenders with scraped documents (processing_metadata.documents_scraped)" +// @Param documents_scraped query bool false "When true, only tenders whose contract_folder_id has documents in MinIO (cached scan)" // @Param limit query int false "Number of items per page (default: 20, max: 100)" // @Param offset query int false "Number of items to skip (default: 0). Prefer cursor for deep pages." // @Param cursor query string false "Opaque cursor from previous page meta.next_cursor" @@ -271,7 +271,7 @@ func (h *TenderHandler) AdminRecommendTenders(c echo.Context) error { // @Param languages query []string false "Filter by languages (comma-separated)" // @Param buyer_organization_id query string false "Filter by buyer organization ID" // @Param notice_publication_id query string false "Filter by TED notice publication ID" -// @Param documents_scraped query bool false "When true, only tenders with scraped documents (processing_metadata.documents_scraped)" +// @Param documents_scraped query bool false "When true, only tenders whose contract_folder_id has documents in MinIO (cached scan)" // @Param limit query int false "Number of items per page (default: 20, max: 100)" // @Param offset query int false "Number of items to skip (default: 0). Prefer cursor for deep pages." // @Param cursor query string false "Opaque cursor from previous page meta.next_cursor" diff --git a/internal/tender/repository.go b/internal/tender/repository.go index fa0712d..e4b006e 100644 --- a/internal/tender/repository.go +++ b/internal/tender/repository.go @@ -13,6 +13,9 @@ import ( "go.mongodb.org/mongo-driver/v2/mongo" ) +// maxContractFolderInClause caps each MongoDB $in batch for documents_scraped MinIO cross-check filters. +const maxContractFolderInClause = 5000 + // TenderRepository interface defines tender data access methods type TenderRepository interface { // Tender CRUD operations @@ -1380,7 +1383,7 @@ func (r *tenderRepository) buildSearchFilter(form *SearchForm) bson.M { } if form.DocumentsScraped { - scrapedClause := bson.M{"processing_metadata.documents_scraped": true} + scrapedClause := documentsScrapedSearchClause(form) if len(filter) == 0 { return scrapedClause } @@ -1399,3 +1402,31 @@ func (r *tenderRepository) buildSearchFilter(form *SearchForm) bson.M { return filter } + +// documentsScrapedSearchClause restricts list queries to tenders whose contract_folder_id +// currently has documents in MinIO (folder IDs are populated by the service layer). +func documentsScrapedSearchClause(form *SearchForm) bson.M { + if form == nil { + return bson.M{"contract_folder_id": bson.M{"$in": []string{}}} + } + return contractFolderIDsSearchClause(form.ContractFolderIDsWithDocuments) +} + +func contractFolderIDsSearchClause(folderIDs []string) bson.M { + if len(folderIDs) == 0 { + return bson.M{"contract_folder_id": bson.M{"$in": []string{}}} + } + if len(folderIDs) <= maxContractFolderInClause { + return bson.M{"contract_folder_id": bson.M{"$in": folderIDs}} + } + + or := make([]bson.M, 0, (len(folderIDs)+maxContractFolderInClause-1)/maxContractFolderInClause) + for i := 0; i < len(folderIDs); i += maxContractFolderInClause { + end := i + maxContractFolderInClause + if end > len(folderIDs) { + end = len(folderIDs) + } + or = append(or, bson.M{"contract_folder_id": bson.M{"$in": folderIDs[i:end]}}) + } + return bson.M{"$or": or} +} diff --git a/internal/tender/repository_search_test.go b/internal/tender/repository_search_test.go index 2c46e1c..39b6ba4 100644 --- a/internal/tender/repository_search_test.go +++ b/internal/tender/repository_search_test.go @@ -2,14 +2,17 @@ package tender import ( "testing" + + "go.mongodb.org/mongo-driver/v2/bson" ) -func TestBuildSearchFilterDocumentsScrapedUsesMongoFlag(t *testing.T) { +func TestBuildSearchFilterDocumentsScrapedWithoutFolderIDsMatchesNothing(t *testing.T) { repo := &tenderRepository{} filter := repo.buildSearchFilter(&SearchForm{DocumentsScraped: true}) - if filter["processing_metadata.documents_scraped"] != true { - t.Fatalf("expected mongo documents_scraped flag, got %#v", filter) + in, ok := filter["contract_folder_id"].(bson.M)["$in"].([]string) + if !ok || len(in) != 0 { + t.Fatalf("expected empty contract_folder_id $in filter, got %#v", filter) } } diff --git a/internal/tender/service.go b/internal/tender/service.go index 4fc3803..b0e00dc 100644 --- a/internal/tender/service.go +++ b/internal/tender/service.go @@ -100,11 +100,15 @@ type tenderService struct { searchListCacheMu sync.Mutex searchListCache map[string]searchListCacheEntry searchListCacheGroup singleflight.Group + + contractFoldersCacheMu sync.Mutex + contractFoldersCache contractFoldersWithDocsCacheEntry + contractFoldersCacheGroup singleflight.Group } const ( - aiSummaryEnrichmentTimeout = 2 * time.Second - aiTranslationEnrichmentTimeout = 2 * time.Second + aiSummaryEnrichmentTimeout = 2 * time.Second + aiTranslationEnrichmentTimeout = 2 * time.Second recommendTranslationEnrichmentTimeout = 5 * time.Second ) @@ -935,6 +939,15 @@ func (s *tenderService) Search(ctx context.Context, form *SearchForm, pagination } func (s *tenderService) searchUncached(ctx context.Context, form *SearchForm, pagination *response.Pagination, language string) (*SearchResponse, error) { + if form.DocumentsScraped { + if err := s.enrichDocumentsScrapedFilter(ctx, form); err != nil { + return nil, err + } + if len(form.ContractFolderIDsWithDocuments) == 0 { + return s.emptySearchResponse(pagination), nil + } + } + page, err := s.repository.Search(ctx, form, pagination) if err != nil { s.logger.Error("Failed to list tenders", map[string]interface{}{ @@ -1759,31 +1772,6 @@ func (s *tenderService) listProceduresWithDocuments(ctx context.Context) ([]ai_s return procedures, nil } -func (s *tenderService) enrichDocumentsScrapedFilter(ctx context.Context, form *SearchForm) error { - procedures, err := s.listProceduresWithDocuments(ctx) - if err != nil { - s.logger.Error("Failed to resolve tenders with scraped documents from MinIO", map[string]interface{}{ - "error": err.Error(), - }) - if errors.Is(err, ai_summarizer.ErrMinIOUnavailable) { - return fmt.Errorf("failed to resolve tenders with scraped documents: MinIO connection unavailable: %w", err) - } - return fmt.Errorf("failed to resolve tenders with scraped documents: %w", err) - } - - folderIDs := make([]string, 0, len(procedures)) - for _, proc := range procedures { - if proc.DocumentCount <= 0 { - continue - } - if id := strings.TrimSpace(proc.ContractFolderID); id != "" { - folderIDs = append(folderIDs, id) - } - } - form.ContractFolderIDsWithDocuments = folderIDs - return nil -} - func (s *tenderService) resolveRecommendedTenders(ctx context.Context, recommendations []company.RecommendedTenderResponse) (map[string]Tender, error) { out := make(map[string]Tender) procRefs := make([]ProcedureReference, 0, len(recommendations))