Enhance AI pipeline service with document metadata synchronization

- Updated the AI pipeline service to include a new `ScrapedDocumentMetadataSyncer` interface for persisting scraped document metadata onto tender records.
- Modified the `NewService` function to accept the new metadata syncer dependency.
- Implemented synchronization of scraped document metadata in the `ScrapeDocuments` and `Run` methods.
- Enhanced the tender service to enrich search filters based on scraped documents and added a new method for syncing scraped documents from storage.
- Updated the `SearchForm` to include `ContractFolderIDsWithDocuments` for better handling of scraped documents in queries.

This update improves the integration of scraped document handling within the AI pipeline, enhancing data consistency and operational efficiency in the tender management system.

cmd/web/main.go

@@ -232,7 +232,7 @@ func main() {
 	documentScraperService := document_scraper.NewService(tenderRepository, logger)
 	dashboardRepository := dashboard.NewRepository(mongoManager, logger)
 	dashboardService := dashboard.NewService(dashboardRepository, logger)
-	aiPipelineService := ai_pipeline.NewService(aiPipelineClient, logger)
+	aiPipelineService := ai_pipeline.NewService(aiPipelineClient, logger, tenderService)
 	logger.Info("Services initialized successfully", map[string]interface{}{
 		"services": []string{"customer", "user", "company", "category", "tender", "feedback", "tender_approval", "inquiry", "flag", "notification", "contact", "cms", "scraper", "kanban", "document_scraper", "dashboard", "ai_pipeline"},
 	})

internal/ai_pipeline/service.go

@@ -46,16 +46,23 @@ type Service interface {
 	GetMinioStats(ctx context.Context) (ai_summarizer.PipelineMinioStatsResponse, error)
 }
 
+// ScrapedDocumentMetadataSyncer persists MinIO scraped-document metadata onto tender records.
+type ScrapedDocumentMetadataSyncer interface {
+	SyncScrapedDocumentsFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) error
+}
+
 type service struct {
-	client Client
+	client         Client
-	logger logger.Logger
+	logger         logger.Logger
+	metadataSyncer ScrapedDocumentMetadataSyncer
 }
 
 // NewService creates an AI pipeline admin service.
-func NewService(client Client, log logger.Logger) Service {
+func NewService(client Client, log logger.Logger, metadataSyncer ScrapedDocumentMetadataSyncer) Service {
 	return &service{
-		client: client,
+		client:         client,
-		logger: log,
+		logger:         log,
+		metadataSyncer: metadataSyncer,
 	}
 }
 
@@ -89,6 +96,10 @@ func (s *service) ScrapeDocuments(ctx context.Context, form ScrapeDocumentsForm)
 		return nil, fmt.Errorf("scrape documents failed: %w", err)
 	}
 
+	if resp != nil && resp.Success && resp.FilesDownloaded > 0 {
+		s.syncScrapedDocumentMetadata(ctx, form.ContractFolderID, form.NoticePublicationID)
+	}
+
 	return resp, nil
 }
 
@@ -222,6 +233,8 @@ func (s *service) Run(ctx context.Context, form PipelineRunForm) (map[string]int
 		return nil, fmt.Errorf("pipeline run failed: %w", err)
 	}
 
+	s.syncScrapedDocumentMetadata(ctx, req.ContractFolderID, req.NoticePublicationID)
+
 	return resp, nil
 }
 
@@ -398,3 +411,16 @@ func validateTenderBatch(forms []TenderRefForm) ([]ai_summarizer.TenderRef, erro
 	}
 	return toTenderRefs(forms), nil
 }
+
+func (s *service) syncScrapedDocumentMetadata(ctx context.Context, contractFolderID, noticePublicationID string) {
+	if s.metadataSyncer == nil {
+		return
+	}
+	if err := s.metadataSyncer.SyncScrapedDocumentsFromStorage(ctx, contractFolderID, noticePublicationID); err != nil {
+		s.logger.Warn("Failed to sync scraped document metadata to Mongo", map[string]interface{}{
+			"contract_folder_id":    contractFolderID,
+			"notice_publication_id": noticePublicationID,
+			"error":                 err.Error(),
+		})
+	}
+}

internal/tender/form.go

@@ -62,6 +62,8 @@ type SearchForm struct {
 	OnlyActiveDeadlines     bool     `query:"-" 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 []string `query:"-" valid:"optional"`
 }
 
 // SearchResponse represents the response for listing tenders

internal/tender/handler.go

@@ -158,7 +158,7 @@ func (h *TenderHandler) Delete(c echo.Context) error {
 // @Param submission_date_to query number false "Submission deadline to (Unix timestamp, seconds)"
 // @Param languages query []string false "Filter by languages (comma-separated)"
 // @Param buyer_organization_id query string false "Filter by buyer organization ID"
-// @Param documents_scraped query bool false "When true, only tenders with scraped documents"
+// @Param documents_scraped query bool false "When true, only tenders whose contract_folder_id has documents in MinIO"
 // @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"
@@ -229,7 +229,7 @@ func (h *TenderHandler) Search(c echo.Context) error {
 // @Param submission_date_to query number false "Submission deadline to (Unix seconds)"
 // @Param languages query []string false "Filter by languages (comma-separated)"
 // @Param buyer_organization_id query string false "Filter by buyer organization ID"
-// @Param documents_scraped query bool false "When true, only tenders with scraped documents"
+// @Param documents_scraped query bool false "When true, only tenders whose contract_folder_id has documents in MinIO"
 // @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"
@@ -312,7 +312,7 @@ func (h *TenderHandler) GetPublicTenders(c echo.Context) error {
 // @Param submission_date_to query number false "Submission deadline to (Unix seconds)"
 // @Param languages query []string false "Filter by languages (comma-separated)"
 // @Param buyer_organization_id query string false "Filter by buyer organization ID"
-// @Param documents_scraped query bool false "When true, only tenders with scraped documents"
+// @Param documents_scraped query bool false "When true, only tenders whose contract_folder_id has documents in MinIO"
 // @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

@@ -1060,11 +1060,10 @@ func (r *tenderRepository) buildSearchFilter(form *SearchForm) bson.M {
 	}
 
 	if form.DocumentsScraped {
-		scrapedClause := bson.M{
+		folderIDs := dedupeNonEmptyStrings(form.ContractFolderIDsWithDocuments)
-			"$or": []bson.M{
+		scrapedClause := bson.M{"contract_folder_id": bson.M{"$in": folderIDs}}
-				{"processing_metadata.documents_scraped": true},
+		if len(folderIDs) == 0 {
-				{"scraped_documents.0": bson.M{"$exists": true}},
+			scrapedClause = bson.M{"_id": bson.M{"$exists": false}}
-			},
 		}
 		if len(filter) == 0 {
 			return scrapedClause

internal/tender/service.go

@@ -70,6 +70,8 @@ type Service interface {
 	// ListTendersWithScrapedDocuments lists tenders that have documents in MinIO. When syncMongoMetadata is true,
 	// each tender is re-listed from MinIO and scraped document metadata is persisted to Mongo (slow).
 	ListTendersWithScrapedDocuments(ctx context.Context, pagination *response.Pagination, syncMongoMetadata bool) (*TendersWithScrapedDocumentsResponse, error)
+	// SyncScrapedDocumentsFromStorage lists documents in MinIO for a tender and persists scraped metadata to Mongo.
+	SyncScrapedDocumentsFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) error
 
 	// Maintenance operations
 	UpdateExpiredTenders(ctx context.Context) error
@@ -886,6 +888,15 @@ func (s *tenderService) Search(ctx context.Context, form *SearchForm, pagination
 		}
 	}
 
+	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{}{
@@ -997,6 +1008,15 @@ func (s *tenderService) Recommend(ctx context.Context, form *SearchForm, paginat
 		}
 	}
 
+	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{}{
@@ -1410,25 +1430,13 @@ func (s *tenderService) GetAllAISummaries(ctx context.Context) ([]ai_summarizer.
 // documents in MinIO. By default uses counts from the initial MinIO procedure scan only (fast).
 // Set syncMongoMetadata=true to list every notice folder again and persist ScrapedDocuments to Mongo (slow).
 func (s *tenderService) ListTendersWithScrapedDocuments(ctx context.Context, pagination *response.Pagination, syncMongoMetadata bool) (*TendersWithScrapedDocumentsResponse, error) {
-	if s.aiSummarizerStorage == nil {
-		return nil, fmt.Errorf("document storage is not configured")
-	}
-
-	type procedureLister interface {
-		ListProceduresWithDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error)
-	}
-	storage, ok := s.aiSummarizerStorage.(procedureLister)
-	if !ok {
-		return nil, fmt.Errorf("document storage does not support procedure listing")
-	}
-
 	s.logger.Info("Listing tenders with scraped documents from MinIO", map[string]interface{}{
 		"limit":               pagination.Limit,
 		"offset":              pagination.Offset,
 		"sync_mongo_metadata": syncMongoMetadata,
 	})
 
-	procedures, err := storage.ListProceduresWithDocuments(ctx)
+	procedures, err := s.listProceduresWithDocuments(ctx)
 	if err != nil {
 		fields := map[string]interface{}{"error": err.Error()}
 		if errors.Is(err, ai_summarizer.ErrMinIOUnavailable) {
@@ -1564,6 +1572,118 @@ func (s *tenderService) persistScrapedDocuments(ctx context.Context, t *Tender,
 	}
 }
 
+type procedureDocumentLister interface {
+	ListProceduresWithDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error)
+}
+
+func (s *tenderService) procedureDocumentStorage() (procedureDocumentLister, error) {
+	if s.aiSummarizerStorage == nil {
+		return nil, fmt.Errorf("document storage is not configured")
+	}
+	storage, ok := s.aiSummarizerStorage.(procedureDocumentLister)
+	if !ok {
+		return nil, fmt.Errorf("document storage does not support procedure listing")
+	}
+	return storage, nil
+}
+
+func (s *tenderService) listProceduresWithDocuments(ctx context.Context) ([]ai_summarizer.ProcedureDocumentsSummary, error) {
+	storage, err := s.procedureDocumentStorage()
+	if err != nil {
+		return nil, err
+	}
+	procedures, err := storage.ListProceduresWithDocuments(ctx)
+	if err != nil {
+		return nil, err
+	}
+	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) emptySearchResponse(pagination *response.Pagination) *SearchResponse {
+	return &SearchResponse{
+		Tenders:  []TenderResponse{},
+		Metadata: pagination.ListMeta(0, "", false, pagination.Offset),
+	}
+}
+
+// SyncScrapedDocumentsFromStorage lists documents in MinIO for a tender and persists scraped metadata to Mongo.
+func (s *tenderService) SyncScrapedDocumentsFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) error {
+	contractFolderID = strings.TrimSpace(contractFolderID)
+	noticePublicationID = strings.TrimSpace(noticePublicationID)
+	if contractFolderID == "" {
+		return fmt.Errorf("contract folder ID is required")
+	}
+
+	var t *Tender
+	var err error
+	if noticePublicationID != "" {
+		t, err = s.repository.GetByNoticePublicationID(ctx, noticePublicationID)
+	} else {
+		byFolder, lookupErr := s.repository.GetLatestByContractFolderIDs(ctx, []string{contractFolderID})
+		err = lookupErr
+		if lookupErr == nil {
+			t = byFolder[contractFolderID]
+		}
+	}
+	if err != nil {
+		s.logger.Error("Failed to load tender for scraped document sync", map[string]interface{}{
+			"contract_folder_id":    contractFolderID,
+			"notice_publication_id": noticePublicationID,
+			"error":                 err.Error(),
+		})
+		return fmt.Errorf("failed to load tender for scraped document sync: %w", err)
+	}
+	if t == nil {
+		s.logger.Warn("Skipping scraped document metadata sync: tender not found", map[string]interface{}{
+			"contract_folder_id":    contractFolderID,
+			"notice_publication_id": noticePublicationID,
+		})
+		return nil
+	}
+
+	docs, err := s.listDocumentsForTender(ctx, t)
+	if err != nil {
+		s.logger.Error("Failed to list scraped documents from storage", map[string]interface{}{
+			"tender_id":             t.ID.Hex(),
+			"contract_folder_id":    contractFolderID,
+			"notice_publication_id": noticePublicationID,
+			"error":                 err.Error(),
+		})
+		return fmt.Errorf("failed to list scraped documents from storage: %w", err)
+	}
+	if len(docs) == 0 {
+		return nil
+	}
+
+	s.persistScrapedDocuments(ctx, t, docs)
+	return nil
+}
+
 // TriggerAITranslate triggers on-demand AI translation for a tender. Results are
 // read from MinIO when available, otherwise fetched from the AI service (which persists to MinIO).
 func (s *tenderService) TriggerAITranslate(ctx context.Context, id, language string) (*AITranslateResponse, error) {