Merge pull request 'Add documents scraped filter functionality and related tests' (#51) from documents-scraped into develop

Reviewed-on: https://repo.ravanertebat.com/TM/tm_back/pulls/51

internal/tender/documents_scraped_filter.go

@@ -0,0 +1,120 @@
+package tender
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"tm/pkg/ai_summarizer"
+)
+
+const (
+	// contractFoldersWithDocsCacheTTL avoids scanning MinIO on every documents_scraped list request.
+	// 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
+)
+
+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()
+
+		// 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)
+		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
+}

internal/tender/documents_scraped_filter_test.go

@@ -0,0 +1,241 @@
+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 interface {
+		ListProceduresWithDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error)
+	}
+}
+
+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)
+	}
+}
+
+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{
+			{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())
+	}
+}

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"`

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"

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}
+}

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 {
+	in, ok := filter["contract_folder_id"].(bson.M)["$in"].([]string)
-		t.Fatalf("expected mongo documents_scraped flag, got %#v", filter)
+	if !ok || len(in) != 0 {
+		t.Fatalf("expected empty contract_folder_id $in filter, got %#v", filter)
 	}
 }
 

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
+	aiSummaryEnrichmentTimeout            = 2 * time.Second
-	aiTranslationEnrichmentTimeout    = 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))