From 5c93e0f01b397d17a581c7ac8000e2ee3db120e0 Mon Sep 17 00:00:00 2001 From: Mazyar Date: Sun, 12 Jul 2026 00:20:39 +0330 Subject: [PATCH 1/3] Add documents scraped filter functionality and related tests - Introduced `documents_scraped_filter.go` to handle filtering tenders based on the presence of documents in MinIO, utilizing a caching mechanism to optimize performance. - Implemented `enrichDocumentsScrapedFilter` and `cachedContractFolderIDsWithDocuments` methods in the `tenderService` for efficient retrieval of contract folder IDs with documents. - Added unit tests in `documents_scraped_filter_test.go` to validate the new filtering logic and caching behavior, ensuring accurate results and error handling. - Updated `SearchForm` documentation to clarify the caching aspect of `ContractFolderIDsWithDocuments`. - Enhanced comments in the handler and repository to reflect the new filtering logic and its implications. This update improves the tender search functionality by efficiently managing document presence checks, leading to better performance and user experience in the tender management system. --- internal/tender/documents_scraped_filter.go | 117 +++++++++++ .../tender/documents_scraped_filter_test.go | 181 ++++++++++++++++++ internal/tender/form.go | 2 +- internal/tender/handler.go | 4 +- internal/tender/repository.go | 33 +++- internal/tender/repository_search_test.go | 9 +- internal/tender/service.go | 42 ++-- 7 files changed, 354 insertions(+), 34 deletions(-) create mode 100644 internal/tender/documents_scraped_filter.go create mode 100644 internal/tender/documents_scraped_filter_test.go 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)) From 0fde5772e29a3bfec7a9018d8bbcee928583a009 Mon Sep 17 00:00:00 2001 From: Mazyar Date: Sun, 12 Jul 2026 00:20:39 +0330 Subject: [PATCH 2/3] Add documents scraped filter functionality and related tests - Introduced `documents_scraped_filter.go` to handle filtering tenders based on the presence of documents in MinIO, utilizing a caching mechanism to optimize performance. - Implemented `enrichDocumentsScrapedFilter` and `cachedContractFolderIDsWithDocuments` methods in the `tenderService` for efficient retrieval of contract folder IDs with documents. - Added unit tests in `documents_scraped_filter_test.go` to validate the new filtering logic and caching behavior, ensuring accurate results and error handling. - Updated `SearchForm` documentation to clarify the caching aspect of `ContractFolderIDsWithDocuments`. - Enhanced comments in the handler and repository to reflect the new filtering logic and its implications. This update improves the tender search functionality by efficiently managing document presence checks, leading to better performance and user experience in the tender management system. --- internal/tender/documents_scraped_filter.go | 117 +++++++++++ .../tender/documents_scraped_filter_test.go | 181 ++++++++++++++++++ internal/tender/form.go | 2 +- internal/tender/handler.go | 4 +- internal/tender/repository.go | 33 +++- internal/tender/repository_search_test.go | 9 +- internal/tender/service.go | 42 ++-- 7 files changed, 354 insertions(+), 34 deletions(-) create mode 100644 internal/tender/documents_scraped_filter.go create mode 100644 internal/tender/documents_scraped_filter_test.go 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)) From 04c410217086659e27ac0cff5e1311911f04599c Mon Sep 17 00:00:00 2001 From: Mazyar Date: Sun, 12 Jul 2026 21:51:59 +0330 Subject: [PATCH 3/3] Enhance documents scraped filter with cancellation handling and caching improvements - Updated `documents_scraped_filter.go` to extend the caching mechanism for contract folder IDs, increasing the cache TTL to 15 minutes to prevent premature invalidation during slow scans. - Modified the `cachedContractFolderIDsWithDocuments` method to detach from the caller's cancellation, ensuring that shared scans are not aborted by client disconnects or timeouts. - Introduced a new test, `TestCachedContractFolderIDsSurvivesCallerCancellation`, in `documents_scraped_filter_test.go` to validate the behavior of the caching mechanism under cancellation scenarios. - Enhanced the `scanStartedProcedureLister` to manage scan initiation and cancellation more effectively, improving the robustness of the document retrieval process. This update improves the reliability and performance of the tender management system's document filtering capabilities, particularly in scenarios with long-running scans. --- internal/tender/documents_scraped_filter.go | 7 ++- .../tender/documents_scraped_filter_test.go | 62 ++++++++++++++++++- 2 files changed, 66 insertions(+), 3 deletions(-) diff --git a/internal/tender/documents_scraped_filter.go b/internal/tender/documents_scraped_filter.go index 8265fbc..899c0cc 100644 --- a/internal/tender/documents_scraped_filter.go +++ b/internal/tender/documents_scraped_filter.go @@ -12,7 +12,8 @@ import ( const ( // contractFoldersWithDocsCacheTTL avoids scanning MinIO on every documents_scraped list request. - contractFoldersWithDocsCacheTTL = 60 * time.Second + // Kept >= resolve timeout so a slow scan is not immediately invalidated for waiters. + contractFoldersWithDocsCacheTTL = 15 * time.Minute // contractFoldersWithDocsResolveTimeout bounds a single MinIO procedure scan. contractFoldersWithDocsResolveTimeout = 3 * time.Minute ) @@ -65,7 +66,9 @@ func (s *tenderService) cachedContractFolderIDsWithDocuments(ctx context.Context } s.contractFoldersCacheMu.Unlock() - resolveCtx, cancel := context.WithTimeout(ctx, contractFoldersWithDocsResolveTimeout) + // Detach from the caller's cancellation so a client disconnect or proxy timeout + // does not abort the shared singleflight scan for every waiter. + resolveCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), contractFoldersWithDocsResolveTimeout) defer cancel() procedures, listErr := s.listProceduresWithDocuments(resolveCtx) diff --git a/internal/tender/documents_scraped_filter_test.go b/internal/tender/documents_scraped_filter_test.go index 33e8eaa..f208593 100644 --- a/internal/tender/documents_scraped_filter_test.go +++ b/internal/tender/documents_scraped_filter_test.go @@ -29,7 +29,9 @@ func (m *mockProcedureDocumentLister) ListProceduresWithDocuments(context.Contex } type stubProcedureDocumentStorage struct { - lister *mockProcedureDocumentLister + lister interface { + ListProceduresWithDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error) + } } func (s stubProcedureDocumentStorage) GetSummaryFromStorage(context.Context, string, string) (string, error) { @@ -156,6 +158,64 @@ func TestContractFolderIDsFromProceduresDedupesAndSkipsEmpty(t *testing.T) { } } +type scanStartedProcedureLister struct { + mockProcedureDocumentLister + started chan struct{} +} + +func (m *scanStartedProcedureLister) ListProceduresWithDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error) { + if m.started != nil { + select { + case <-m.started: + default: + close(m.started) + } + } + time.Sleep(50 * time.Millisecond) + if err := ctx.Err(); err != nil { + return nil, err + } + return m.mockProcedureDocumentLister.ListProceduresWithDocuments(ctx) +} + +func TestCachedContractFolderIDsSurvivesCallerCancellation(t *testing.T) { + started := make(chan struct{}) + lister := &scanStartedProcedureLister{ + mockProcedureDocumentLister: mockProcedureDocumentLister{ + procedures: []ai_summarizer.ProcedureDocumentsSummary{ + {ContractFolderID: "folder-1", DocumentCount: 1}, + }, + }, + started: started, + } + svc := &tenderService{aiSummarizerStorage: stubProcedureDocumentStorage{lister: lister}} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + errCh := make(chan error, 1) + var ids []string + go func() { + var err error + ids, err = svc.cachedContractFolderIDsWithDocuments(ctx) + errCh <- err + }() + + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("MinIO scan did not start") + } + cancel() + + if err := <-errCh; err != nil { + t.Fatalf("expected shared scan to finish despite caller cancel: %v", err) + } + if len(ids) != 1 || ids[0] != "folder-1" { + t.Fatalf("unexpected folder ids: %#v", ids) + } +} + func TestContractFoldersWithDocsCacheExpires(t *testing.T) { lister := &mockProcedureDocumentLister{ procedures: []ai_summarizer.ProcedureDocumentsSummary{