From 7d383c36c359fae2eb6b1061bffd0ba6c7a064dd Mon Sep 17 00:00:00 2001 From: Mazyar Date: Mon, 8 Jun 2026 18:01:15 +0330 Subject: [PATCH 1/3] Implement AI analysis trigger endpoint and refactor related components - Added a new endpoint to trigger on-demand agentic analysis for tenders via the AI service. - Introduced `TriggerAIAnalyze` method in the `TenderHandler` to handle requests and responses for AI analysis. - Updated the `tenderService` to include `TriggerAIAnalyze` method, which validates input and interacts with the AI summarizer client. - Enhanced the `AIAnalyzeResponse` and `AIAnalyzeDocument` structures to support the new analysis feature. - Refactored the `DocumentSummarizationWorker` and `TranslationWorker` to remove deprecated MinIO dependencies, streamlining the AI service interactions. This update improves the functionality of the tender management system by allowing users to trigger AI analysis on-demand, enhancing the overall user experience and system capabilities. --- cmd/web/router/routes.go | 1 + cmd/worker/bootstrap/bootstrap.go | 115 +------------- cmd/worker/main.go | 8 +- cmd/worker/workers/summarizer.go | 251 +++++++++--------------------- cmd/worker/workers/translation.go | 145 +++++------------ internal/tender/form.go | 20 +++ internal/tender/handler.go | 32 +++- internal/tender/service.go | 187 ++++++++++++++++++---- pkg/ai_summarizer/client.go | 112 +++++++++++++ pkg/ai_summarizer/entities.go | 140 +++++++++-------- pkg/ai_summarizer/storage.go | 50 +++++- 11 files changed, 559 insertions(+), 502 deletions(-) diff --git a/cmd/web/router/routes.go b/cmd/web/router/routes.go index f76826a..78625b8 100644 --- a/cmd/web/router/routes.go +++ b/cmd/web/router/routes.go @@ -121,6 +121,7 @@ func RegisterAdminRoutes(e *echo.Echo, userHandler *user.Handler, companyHandler tendersGP.GET("/:id/documents/download", tenderHandler.DownloadDocuments) tendersGP.GET("/:id/ai-summary", tenderHandler.GetAISummary) tendersGP.POST("/:id/ai-summarize", tenderHandler.TriggerAISummarize) + tendersGP.POST("/:id/ai-analyze", tenderHandler.TriggerAIAnalyze) tendersGP.POST("/:id/ai-translate", tenderHandler.TriggerAITranslate) } diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index eda46b9..21012c5 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -10,11 +10,9 @@ import ( ai_summarizer "tm/pkg/ai_summarizer" "tm/pkg/config" "tm/pkg/logger" - "tm/pkg/minio" "tm/pkg/mongo" "tm/pkg/notification" "tm/pkg/ollama" - "tm/pkg/queue" "tm/pkg/schedule" ) @@ -78,7 +76,7 @@ func InitMongoDB(conf config.MongoConfig, log logger.Logger) *mongo.ConnectionMa } // Init Worker -func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, minioService *minio.Service, aiClient *ai_summarizer.Client, aiStorage *ai_summarizer.StorageClient) *schedule.CronScheduler { +func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, aiClient *ai_summarizer.Client, aiStorage *ai_summarizer.StorageClient) *schedule.CronScheduler { // Debug: Log worker config appLogger.Info("Worker configuration", map[string]interface{}{ "worker_interval": config.Worker.Interval, @@ -97,70 +95,24 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger // Create a single shared cron scheduler for all recurring jobs scheduler := schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo) - var queueService queue.Queue if config.Queue.Enabled { - var err error - queueConfig := queue.QueueConfig{ - RedisURL: config.Queue.RedisURL, - MaxRetries: config.Queue.MaxRetries, - RetryBackoff: config.Queue.RetryBackoff, - JobTimeout: config.Queue.JobTimeout, - PoolSize: config.Queue.PoolSize, - IsDLQEnabled: config.Queue.IsDLQEnabled, - MaxQueueLength: 0, // Unlimited - } - - queueService, err = queue.NewRedisQueue(queueConfig, appLogger) - if err != nil { - appLogger.Error("Failed to initialize queue service", map[string]interface{}{ - "error": err.Error(), - }) - appLogger.Warn("Queue-based scraping is disabled, falling back to scheduled scraping", map[string]interface{}{}) - config.Queue.Enabled = false - } else { - appLogger.Info("Queue service initialized successfully", map[string]interface{}{ - "mode": config.Queue.Mode, - }) - } - } - - // Start Queue Producer Worker (enqueues unscraped tenders periodically) - if config.Queue.Enabled && (config.Queue.Mode == "producer" || config.Queue.Mode == "both") { - appLogger.Info("Scheduling queue producer worker", map[string]interface{}{ - "batch_size": config.Queue.ProducerBatch, - "interval": config.Queue.ProducerInterval, - }) - - scheduler.AddJob(schedule.Job{ - Name: "Queue Producer Worker Job", - Func: func() { - producer := workers.NewQueueProducerWorker( - mongoManager, - appLogger, - tenderRepo, - queueService, - config.Queue.ProducerBatch, - ) - producer.Run() - }, - Expr: config.Queue.ProducerInterval, + appLogger.Warn("Redis document-scraper queue is deprecated; document scraping is owned by the AI service pipeline", map[string]interface{}{ + "queue_mode": config.Queue.Mode, }) } - // Document summarization via external AI service (requires MinIO + AI_SUMMARIZER_API_BASE_URL) - if minioService != nil && aiClient != nil { + // Document summarization via external AI service (requires AI_SUMMARIZER_API_BASE_URL) + if aiClient != nil { scheduler.AddJob(schedule.Job{ Name: "Document Summarization Worker Job", Func: func() { - worker := workers.NewDocumentSummarizationWorker(mongoManager, appLogger, tenderRepo, minioService, aiClient) + worker := workers.NewDocumentSummarizationWorker(mongoManager, appLogger, tenderRepo, aiClient, aiStorage) worker.Run() }, - Expr: "0 11 * * * *", // Run at 11 AM every day (1 hour after document scraping) + Expr: "0 11 * * * *", // Run at 11 AM every day (after AI pipeline scrape) }) - } else if minioService != nil { - appLogger.Warn("Document summarization worker not scheduled: AI summarizer API is not configured", map[string]interface{}{}) } else { - appLogger.Warn("MinIO service not available, document summarization will be skipped", map[string]interface{}{}) + appLogger.Warn("Document summarization worker not scheduled: AI summarizer API is not configured", map[string]interface{}{}) } // Schedule Notice Worker job (no blocking run at startup; first tick runs on schedule) @@ -395,54 +347,3 @@ func InitOllamaService(conf config.OllamaConfig, log logger.Logger) *ollama.SDK return ollamaSDK } - -// Init MinIO Service -func InitMinIOService(conf config.MinIOConfig, log logger.Logger) *minio.Service { - // Convert bootstrap config to minio config - minioConfig := minio.Config{ - Endpoint: conf.Endpoint, - AccessKeyID: conf.AccessKeyID, - SecretAccessKey: conf.SecretAccessKey, - UseSSL: conf.UseSSL, - Region: conf.Region, - DefaultBucket: conf.DefaultBucket, - HierarchicalBucket: conf.HierarchicalBucket, - } - - // Validate config - if err := minioConfig.Validate(); err != nil { - log.Warn("Invalid MinIO configuration, MinIO features will be disabled", map[string]interface{}{ - "error": err.Error(), - }) - return nil // Return nil instead of panicking - } - - // Create connection manager - connManager, err := minio.NewConnectionManager(minioConfig, log) - if err != nil { - log.Error("Failed to create MinIO connection manager", map[string]interface{}{ - "error": err.Error(), - }) - panic(fmt.Sprintf("Failed to create MinIO connection manager: %v", err)) - } - - // Connect to MinIO - if err := connManager.Connect(); err != nil { - log.Warn("Failed to connect to MinIO, MinIO features will be disabled", map[string]interface{}{ - "endpoint": minioConfig.Endpoint, - "error": err.Error(), - }) - return nil // Return nil instead of panicking - } - - // Create service - service := minio.NewService(connManager, minioConfig, log) - - log.Info("MinIO service initialized successfully", map[string]interface{}{ - "endpoint": conf.Endpoint, - "default_bucket": conf.DefaultBucket, - "hierarchical_bucket": conf.HierarchicalBucket, - }) - - return service -} diff --git a/cmd/worker/main.go b/cmd/worker/main.go index a401a5e..2b60f27 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -37,18 +37,12 @@ func main() { // Initialize notification service notificationService := bootstrap.InitNotificationService(config.Notification, appLogger) - // Initialize MinIO service - minioService := bootstrap.InitMinIOService(config.MinIO, appLogger) - if minioService != nil { - appLogger.Info("MinIO service initialized successfully", map[string]interface{}{}) - } - // Initialize AI summarizer client (translation + document summarization workers) aiSummarizerClient := bootstrap.InitAISummarizerClient(config.AISummarizer, mongoManager, appLogger) aiSummarizerStorage := bootstrap.InitAISummarizerStorage(config.AISummarizer, appLogger) // Initialize Worker - scheduler := bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, minioService, aiSummarizerClient, aiSummarizerStorage) + scheduler := bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, aiSummarizerClient, aiSummarizerStorage) // Set up signal handling for graceful shutdown signalChan := make(chan os.Signal, 1) diff --git a/cmd/worker/workers/summarizer.go b/cmd/worker/workers/summarizer.go index 991d917..51f80f7 100644 --- a/cmd/worker/workers/summarizer.go +++ b/cmd/worker/workers/summarizer.go @@ -2,35 +2,39 @@ package workers import ( "context" - "path/filepath" "strings" "time" "tm/internal/tender" ai_summarizer "tm/pkg/ai_summarizer" "tm/pkg/logger" - "tm/pkg/minio" "tm/pkg/mongo" ) -const presignedDocumentURLSeconds int64 = 7200 - -// DocumentSummarizationWorker asks the external AI service to summarize scraped tender documents. +// DocumentSummarizationWorker triggers AI summarization for scraped tenders. +// The AI service owns MinIO artefacts; this worker marks Mongo processing flags +// after the pipeline or on-demand summarize API completes. type DocumentSummarizationWorker struct { Mongo *mongo.ConnectionManager Logger logger.Logger TenderRepo tender.TenderRepository - MinioSDK *minio.Service AIClient *ai_summarizer.Client + AIStorage *ai_summarizer.StorageClient } // NewDocumentSummarizationWorker creates a document summarization worker. -func NewDocumentSummarizationWorker(mongo *mongo.ConnectionManager, logger logger.Logger, tenderRepo tender.TenderRepository, minioSDK *minio.Service, aiClient *ai_summarizer.Client) *DocumentSummarizationWorker { +func NewDocumentSummarizationWorker( + mongo *mongo.ConnectionManager, + logger logger.Logger, + tenderRepo tender.TenderRepository, + aiClient *ai_summarizer.Client, + aiStorage *ai_summarizer.StorageClient, +) *DocumentSummarizationWorker { return &DocumentSummarizationWorker{ Mongo: mongo, Logger: logger, TenderRepo: tenderRepo, - MinioSDK: minioSDK, AIClient: aiClient, + AIStorage: aiStorage, } } @@ -42,9 +46,11 @@ func (w *DocumentSummarizationWorker) Run() { w.Logger.Warn("AI summarizer client not configured, skipping document summarization run", map[string]interface{}{}) return } - if w.MinioSDK == nil { - w.Logger.Warn("MinIO not configured, skipping document summarization run", map[string]interface{}{}) - return + + if _, err := w.AIClient.TriggerPipelineSummarize(context.Background()); err != nil { + w.Logger.Warn("Failed to trigger AI pipeline summarize (continuing with per-tender sync)", map[string]interface{}{ + "error": err.Error(), + }) } limit := 5 @@ -54,7 +60,7 @@ func (w *DocumentSummarizationWorker) Run() { w.Logger.Error("Failed to get un-summarized tenders", map[string]interface{}{ "error": err.Error(), }) - continue + return } w.Logger.Info("Found tenders for document summarization", map[string]interface{}{ @@ -63,7 +69,6 @@ func (w *DocumentSummarizationWorker) Run() { }) if len(tenders) == 0 { - w.Logger.Info("No more tenders to process for document summarization", map[string]interface{}{}) break } @@ -72,78 +77,78 @@ func (w *DocumentSummarizationWorker) Run() { w.Logger.Info("Processing tender for document summarization", map[string]interface{}{ "tender_id": t.ID.Hex(), "notice_publication_id": t.NoticePublicationID, - "document_count": len(t.ScrapedDocuments), }) - w.summarizeDocumentsForTender(t) + w.summarizeTender(t) } } w.Logger.Info("Document summarization worker completed", map[string]interface{}{}) } -func (w *DocumentSummarizationWorker) summarizeDocumentsForTender(t *tender.Tender) { - if len(t.ScrapedDocuments) == 0 { - w.Logger.Warn("No scraped documents found for tender", map[string]interface{}{ - "tender_id": t.ID.Hex(), - "notice_publication_id": t.NoticePublicationID, +func (w *DocumentSummarizationWorker) summarizeTender(t *tender.Tender) { + if strings.TrimSpace(t.NoticePublicationID) == "" { + w.Logger.Warn("Skipping tender summarization: missing notice_publication_id", map[string]interface{}{ + "tender_id": t.ID.Hex(), + }) + return + } + if strings.TrimSpace(t.ContractFolderID) == "" { + w.Logger.Warn("Skipping tender summarization: missing contract_folder_id", map[string]interface{}{ + "tender_id": t.ID.Hex(), }) - t.ProcessingMetadata.DocumentsSummarized = true - t.ProcessingMetadata.DocumentsSummarizedAt = time.Now().Unix() - _ = w.updateTenderSummarizationStatus(t) return } - summaries := make([]tender.DocumentSummary, 0, len(t.ScrapedDocuments)) - summarizedAt := time.Now().Unix() + ctx := context.Background() + if w.isSummaryReadyInStorage(ctx, t) { + w.markTenderSummarized(t) + return + } - for _, doc := range t.ScrapedDocuments { - summaryText, model, errStr := w.summarizeDocumentViaAI(context.Background(), t, doc) - if errStr != "" { - w.Logger.Error("Failed to summarize document", map[string]interface{}{ - "tender_id": t.ID.Hex(), - "notice_publication_id": t.NoticePublicationID, - "document_name": doc.Filename, - "object_name": doc.ObjectName, - "error": errStr, - }) - summaries = append(summaries, tender.DocumentSummary{ - DocumentName: doc.Filename, - ObjectName: doc.ObjectName, - BucketName: doc.BucketName, - Summary: "", - SummaryLanguage: "en", - DocumentType: w.getDocumentType(doc.Filename), - SummarizedAt: summarizedAt, - SummaryModel: model, - Error: errStr, - }) - continue - } - - summaries = append(summaries, tender.DocumentSummary{ - DocumentName: doc.Filename, - ObjectName: doc.ObjectName, - BucketName: doc.BucketName, - Summary: summaryText, - SummaryLanguage: "en", - DocumentType: w.getDocumentType(doc.Filename), - SummarizedAt: summarizedAt, - SummaryModel: model, - }) - - w.Logger.Info("Document summarized successfully", map[string]interface{}{ + req := ai_summarizer.NewSummarizeRequest(t.ContractFolderID, t.NoticePublicationID) + if _, err := w.AIClient.FetchSummaryOnDemand(ctx, req); err != nil { + w.Logger.Error("Failed to summarize tender via AI service", map[string]interface{}{ + "tender_id": t.ID.Hex(), + "notice_publication_id": t.NoticePublicationID, + "contract_folder_id": t.ContractFolderID, + "error": err.Error(), + }) + return + } + + if !w.isSummaryReadyInStorage(ctx, t) { + w.Logger.Warn("AI summarize completed but storage is not ready yet", map[string]interface{}{ "tender_id": t.ID.Hex(), "notice_publication_id": t.NoticePublicationID, - "document_name": doc.Filename, - "summary_length": len(summaryText), }) } - t.DocumentSummaries = summaries - t.ProcessingMetadata.DocumentsSummarized = true - t.ProcessingMetadata.DocumentsSummarizedAt = summarizedAt + w.markTenderSummarized(t) +} - if err := w.updateTenderSummarizationStatus(t); err != nil { +func (w *DocumentSummarizationWorker) isSummaryReadyInStorage(ctx context.Context, t *tender.Tender) bool { + if w.AIStorage == nil { + return false + } + ready, err := w.AIStorage.IsSummaryReady(ctx, t.ContractFolderID, t.NoticePublicationID) + if err != nil { + w.Logger.Debug("Could not check summarization status in storage", map[string]interface{}{ + "tender_id": t.ID.Hex(), + "notice_publication_id": t.NoticePublicationID, + "error": err.Error(), + }) + return false + } + return ready +} + +func (w *DocumentSummarizationWorker) markTenderSummarized(t *tender.Tender) { + now := time.Now().Unix() + t.ProcessingMetadata.DocumentsSummarized = true + t.ProcessingMetadata.DocumentsSummarizedAt = now + t.UpdatedAt = now + + if err := w.TenderRepo.Update(context.Background(), t); err != nil { w.Logger.Error("Failed to update tender summarization status", map[string]interface{}{ "tender_id": t.ID.Hex(), "notice_publication_id": t.NoticePublicationID, @@ -152,118 +157,8 @@ func (w *DocumentSummarizationWorker) summarizeDocumentsForTender(t *tender.Tend return } - w.Logger.Info("Document summarization completed for tender", map[string]interface{}{ + w.Logger.Info("Tender marked as summarized", map[string]interface{}{ "tender_id": t.ID.Hex(), "notice_publication_id": t.NoticePublicationID, - "summarized_count": len(summaries), - "total_documents": len(t.ScrapedDocuments), }) } - -func (w *DocumentSummarizationWorker) summarizeDocumentViaAI(ctx context.Context, t *tender.Tender, doc tender.ScrapedDocument) (summaryText, model, errStr string) { - docURL, err := w.MinioSDK.GetPresignedURL(ctx, doc.BucketName, doc.ObjectName, presignedDocumentURLSeconds) - if err != nil { - return "", "", err.Error() - } - - if strings.TrimSpace(t.NoticePublicationID) == "" { - return "", "", "tender has no notice_publication_id for AI summarize request" - } - if strings.TrimSpace(t.ContractFolderID) == "" { - return "", "", "tender has no contract_folder_id for AI summarize request" - } - - req := ai_summarizer.NewSummarizeRequest(ai_summarizer.SummarizeRequestInput{ - ContractFolderID: t.ContractFolderID, - NoticePublicationID: t.NoticePublicationID, - DocumentURL: docURL, - Title: t.Title, - Description: t.Description, - Country: t.CountryCode, - Currency: t.Currency, - SubmissionDeadline: t.SubmissionDeadline, - EstimatedValue: t.EstimatedValue, - AuthorityName: authorityNameFromTender(t), - }) - - resp, err := w.AIClient.FetchSummaryOnDemand(ctx, req) - if err != nil { - return "", "", err.Error() - } - if resp == nil { - return "", "", "AI summarize returned nil response" - } - - summaryText, model = pickDocumentSummary(resp.Summaries, doc.Filename) - if summaryText == "" && len(resp.Summaries) > 0 { - // Fall back to first summary if filename did not match - summaryText = resp.Summaries[0].Summary - model = resp.Summaries[0].SummaryModel - } - if summaryText == "" { - return "", model, "AI summarize returned no summary text for document" - } - return summaryText, model, "" -} - -func authorityNameFromTender(t *tender.Tender) string { - if t == nil || t.BuyerOrganization == nil { - return "" - } - return strings.TrimSpace(t.BuyerOrganization.Name) -} - -func pickDocumentSummary(items []ai_summarizer.DocumentSummary, filename string) (text, model string) { - filename = strings.TrimSpace(filename) - for _, s := range items { - if strings.EqualFold(strings.TrimSpace(s.DocumentName), filename) { - return strings.TrimSpace(s.Summary), strings.TrimSpace(s.SummaryModel) - } - } - for _, s := range items { - if strings.TrimSpace(s.Summary) != "" { - return strings.TrimSpace(s.Summary), strings.TrimSpace(s.SummaryModel) - } - } - return "", "" -} - -func (w *DocumentSummarizationWorker) getDocumentType(filename string) string { - ext := strings.ToLower(filepath.Ext(filename)) - switch ext { - case ".xlsx": - return "xlsx" - case ".pdf": - return "pdf" - case ".docx": - return "docx" - case ".doc": - return "doc" - case ".txt": - return "txt" - case ".html", ".htm": - return "html" - default: - return "unknown" - } -} - -func (w *DocumentSummarizationWorker) updateTenderSummarizationStatus(tender *tender.Tender) error { - err := w.TenderRepo.Update(context.Background(), tender) - if err != nil { - w.Logger.Error("Failed to update tender summarization status", map[string]interface{}{ - "tender_id": tender.ID.Hex(), - "error": err.Error(), - }) - return err - } - - w.Logger.Debug("Updated tender summarization status", map[string]interface{}{ - "tender_id": tender.ID.Hex(), - "documents_summarized": tender.ProcessingMetadata.DocumentsSummarized, - "documents_summarized_at": tender.ProcessingMetadata.DocumentsSummarizedAt, - "summary_count": len(tender.DocumentSummaries), - }) - - return nil -} diff --git a/cmd/worker/workers/translation.go b/cmd/worker/workers/translation.go index bb8e6a5..9f94f63 100644 --- a/cmd/worker/workers/translation.go +++ b/cmd/worker/workers/translation.go @@ -2,7 +2,6 @@ package workers import ( "context" - "errors" "strings" "tm/internal/tender" "tm/pkg/ai_summarizer" @@ -10,17 +9,18 @@ import ( "tm/pkg/mongo" ) -// TranslationWorker backfills missing tender translations using the AI pipeline -// and on-demand translate API. Results are persisted by the AI service in MinIO only. +// TranslationWorker syncs translation completion from the AI pipeline MinIO storage. +// Batch translation is triggered via POST /pipeline/translate; this worker only +// verifies which tenders now have translations available in storage. type TranslationWorker struct { - Mongo *mongo.ConnectionManager - Logger logger.Logger - TenderRepo tender.TenderRepository - AIClient *ai_summarizer.Client - AIStorage *ai_summarizer.StorageClient + Mongo *mongo.ConnectionManager + Logger logger.Logger + TenderRepo tender.TenderRepository + AIClient *ai_summarizer.Client + AIStorage *ai_summarizer.StorageClient TranslationSuccessCounter *mongo.Counter - TargetLanguages []string - BatchSize int + TargetLanguages []string + BatchSize int } // NewTranslationWorker creates a translation worker for the given target languages. @@ -66,7 +66,7 @@ func NewTranslationWorker( } } -// Run starts backfilling translations for tenders missing target-language content. +// Run triggers the AI translation pipeline and logs tenders with ready translations in MinIO. func (w *TranslationWorker) Run() { if w.AIClient == nil { w.Logger.Warn("Translation worker skipped: AI client is not configured", map[string]interface{}{ @@ -83,14 +83,15 @@ func (w *TranslationWorker) Run() { startCount := w.getSuccessfulTranslationCount(mongo.AITranslationSuccessCounterKeyDailyJob) if _, err := w.AIClient.TriggerPipelineTranslate(context.Background(), w.TargetLanguages); err != nil { - w.Logger.Warn("Failed to trigger AI pipeline translate (continuing with per-tender sync)", map[string]interface{}{ + w.Logger.Warn("Failed to trigger AI pipeline translate", map[string]interface{}{ "target_languages": w.TargetLanguages, "error": err.Error(), }) } + readyCount := 0 for _, language := range w.TargetLanguages { - w.runForLanguage(language) + readyCount += w.syncLanguageFromStorage(language) } endCount := w.getSuccessfulTranslationCount(mongo.AITranslationSuccessCounterKeyDailyJob) @@ -101,10 +102,11 @@ func (w *TranslationWorker) Run() { } w.Logger.Info("Translation worker completed", map[string]interface{}{ - "target_languages": w.TargetLanguages, - "successful_ai_translation_requests_run": runCount, - "successful_ai_translation_requests_daily_job": endCount, - "successful_ai_translation_requests_total": totalCount, + "target_languages": w.TargetLanguages, + "translations_ready_in_storage": readyCount, + "successful_ai_translation_requests_run": runCount, + "successful_ai_translation_requests_daily_job": endCount, + "successful_ai_translation_requests_total": totalCount, }) } @@ -123,19 +125,27 @@ func (w *TranslationWorker) getSuccessfulTranslationCount(key string) int64 { return count } -func (w *TranslationWorker) runForLanguage(language string) { +func (w *TranslationWorker) syncLanguageFromStorage(language string) int { + if w.AIStorage == nil { + w.Logger.Warn("Translation storage client not configured; skipping MinIO sync", map[string]interface{}{ + "target_language": language, + }) + return 0 + } + + readyCount := 0 skip := 0 for { tenders, totalCount, err := w.TenderRepo.GetUnTranslatedTenders(context.Background(), language, w.BatchSize, skip) if err != nil { - w.Logger.Error("Failed to fetch tenders for translation backfill", map[string]interface{}{ + w.Logger.Error("Failed to fetch tenders for translation sync", map[string]interface{}{ "target_language": language, "error": err.Error(), }) - return + return readyCount } - w.Logger.Info("Fetched tenders for translation backfill", map[string]interface{}{ + w.Logger.Info("Fetched tenders for translation sync", map[string]interface{}{ "target_language": language, "batch_count": len(tenders), "total_count": totalCount, @@ -143,63 +153,29 @@ func (w *TranslationWorker) runForLanguage(language string) { }) if len(tenders) == 0 { - return + return readyCount } for _, t := range tenders { - w.translateTender(&t, language) + if w.hasTranslationInStorage(&t, language) { + readyCount++ + w.Logger.Debug("Translation available in MinIO", map[string]interface{}{ + "tender_id": t.ID.Hex(), + "notice_id": t.NoticePublicationID, + "target_language": language, + }) + } } skip += len(tenders) if int64(skip) >= totalCount { - return + return readyCount } } } -func (w *TranslationWorker) translateTender(t *tender.Tender, language string) { - if t.NoticePublicationID == "" { - return - } - if strings.TrimSpace(t.ContractFolderID) == "" { - w.Logger.Debug("Skipping tender translation: missing contract_folder_id", map[string]interface{}{ - "tender_id": t.ID.Hex(), - }) - return - } - if w.hasTranslationInStorage(t, language) { - w.Logger.Debug("Skipping tender translation: already available in MinIO", map[string]interface{}{ - "tender_id": t.ID.Hex(), - "notice_id": t.NoticePublicationID, - "target_language": language, - }) - return - } - - title, description, source, err := w.resolveTranslation(t, language) - if err != nil { - w.Logger.Error("Failed to translate tender", map[string]interface{}{ - "tender_id": t.ID.Hex(), - "notice_id": t.NoticePublicationID, - "target_language": language, - "translation_source": source, - "translation_error": err.Error(), - }) - return - } - - w.Logger.Info("Tender translated successfully", map[string]interface{}{ - "tender_id": t.ID.Hex(), - "notice_id": t.NoticePublicationID, - "target_language": language, - "translation_source": source, - "title_len": len(title), - "description_len": len(description), - }) -} - func (w *TranslationWorker) hasTranslationInStorage(t *tender.Tender, language string) bool { - if w.AIStorage == nil { + if strings.TrimSpace(t.ContractFolderID) == "" || t.NoticePublicationID == "" { return false } stored, err := w.AIStorage.GetTranslationFromStorage( @@ -210,40 +186,3 @@ func (w *TranslationWorker) hasTranslationInStorage(t *tender.Tender, language s ) return err == nil && strings.TrimSpace(stored.Title) != "" } - -func (w *TranslationWorker) resolveTranslation(t *tender.Tender, language string) (title, description, source string, err error) { - if w.AIStorage != nil { - stored, storageErr := w.AIStorage.GetTranslationFromStorage( - context.Background(), - t.ContractFolderID, - t.NoticePublicationID, - language, - ) - if storageErr == nil { - return stored.Title, stored.Description, "storage", nil - } - if !errors.Is(storageErr, ai_summarizer.ErrTranslationNotReady) && - !errors.Is(storageErr, ai_summarizer.ErrObjectNotFound) && - !errors.Is(storageErr, ai_summarizer.ErrNoTranslation) { - w.Logger.Debug("Translation not available from storage yet", map[string]interface{}{ - "tender_id": t.ID.Hex(), - "notice_id": t.NoticePublicationID, - "target_language": language, - "error": storageErr.Error(), - }) - } - } - - resp, apiErr := w.AIClient.FetchTranslationOnDemand(context.Background(), ai_summarizer.TranslateRequest{ - ContractFolderID: t.ContractFolderID, - NoticePublicationID: t.NoticePublicationID, - Title: t.Title, - Description: t.Description, - Language: language, - RequestSource: ai_summarizer.TranslationRequestSourceDailyJob, - }) - if apiErr != nil { - return "", "", "api", apiErr - } - return resp.TranslatedTitle, resp.TranslatedDescription, "api", nil -} diff --git a/internal/tender/form.go b/internal/tender/form.go index 4a2bfaa..8cd8f23 100644 --- a/internal/tender/form.go +++ b/internal/tender/form.go @@ -573,3 +573,23 @@ type AIDocumentSummary struct { SummaryModel string `json:"summary_model"` Error string `json:"error,omitempty"` } + +// AIAnalyzeResponse represents the response for on-demand agentic analysis. +type AIAnalyzeResponse struct { + ContractFolderID string `json:"contract_folder_id"` + NoticePublicationID string `json:"notice_publication_id"` + Summary string `json:"summary"` + Documents []AIAnalyzeDocument `json:"documents"` + ToolCallsLog []string `json:"tool_calls_log"` + Iterations int `json:"iterations"` + Model string `json:"model"` +} + +// AIAnalyzeDocument describes one document referenced during analysis. +type AIAnalyzeDocument struct { + Filename string `json:"filename"` + SizeBytes int64 `json:"size_bytes"` + TimesRead int `json:"times_read"` + TimesQueried int `json:"times_queried"` + Description string `json:"description"` +} diff --git a/internal/tender/handler.go b/internal/tender/handler.go index 83863de..c239f1f 100644 --- a/internal/tender/handler.go +++ b/internal/tender/handler.go @@ -398,7 +398,7 @@ func (h *TenderHandler) GetPublicTenderDetails(c echo.Context) error { // GetDocumentSummaries retrieves document summaries for a tender // @Summary Get document summaries for a tender -// @Description Retrieve AI-generated summaries of documents for a specific tender +// @Description Retrieve per-document AI summaries from the AI service MinIO storage for a specific tender // @Tags Tenders // @Produce json // @Param id path string true "Tender ID" @@ -595,6 +595,36 @@ func (h *TenderHandler) TriggerAISummarize(c echo.Context) error { return response.Success(c, result, "AI summarization completed successfully") } +// TriggerAIAnalyze triggers on-demand agentic analysis for a tender +// @Summary Trigger AI analysis +// @Description Trigger on-demand agentic document analysis for a tender via the external AI service +// @Tags Admin-Tenders +// @Produce json +// @Param id path string true "Tender ID" +// @Success 200 {object} response.APIResponse{data=AIAnalyzeResponse} +// @Failure 400 {object} response.APIResponse +// @Failure 404 {object} response.APIResponse +// @Failure 500 {object} response.APIResponse +// @Router /admin/v1/tenders/{id}/ai-analyze [post] +// @Security BearerAuth +func (h *TenderHandler) TriggerAIAnalyze(c echo.Context) error { + id := c.Param("id") + if id == "" { + return response.BadRequest(c, "Tender ID is required", "ID parameter cannot be empty") + } + + result, err := h.service.TriggerAIAnalyze(c.Request().Context(), id) + if err != nil { + h.logger.Error("Failed to trigger AI analysis", map[string]interface{}{ + "tender_id": id, + "error": err.Error(), + }) + return response.InternalServerError(c, err.Error()) + } + + return response.Success(c, result, "AI analysis completed successfully") +} + // TriggerAITranslate triggers on-demand AI translation for a tender // @Summary Trigger AI translation // @Description Trigger on-demand AI translation for tender title and description (cached in MinIO by the AI service) diff --git a/internal/tender/service.go b/internal/tender/service.go index a3e7c02..44cd986 100644 --- a/internal/tender/service.go +++ b/internal/tender/service.go @@ -18,10 +18,11 @@ import ( "tm/pkg/response" ) -// AISummarizerClient defines the interface for on-demand AI summarization +// AISummarizerClient defines the interface for on-demand AI operations. type AISummarizerClient interface { FetchSummaryOnDemand(ctx context.Context, reqBody ai_summarizer.SummarizeRequest) (*ai_summarizer.SummarizeResponse, error) FetchTranslationOnDemand(ctx context.Context, reqBody ai_summarizer.TranslateRequest) (*ai_summarizer.TranslateResponse, error) + FetchAnalyzeOnDemand(ctx context.Context, reqBody ai_summarizer.AnalyzeRequest) (*ai_summarizer.AnalyzeResponse, error) TriggerPipelineTranslate(ctx context.Context, languages []string) (*ai_summarizer.PipelineTranslateResponse, error) } @@ -30,6 +31,7 @@ type AISummarizerClient interface { // PROC_// in the shared bucket. type AISummarizerStorage interface { GetSummaryFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) (string, error) + GetDocumentSummariesFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) ([]ai_summarizer.DocumentSummary, error) GetTranslationFromStorage(ctx context.Context, contractFolderID, noticePublicationID, language string) (ai_summarizer.StoredTranslation, error) ListDocuments(ctx context.Context, contractFolderID, noticePublicationID string) ([]ai_summarizer.StoredDocument, error) DownloadDocument(ctx context.Context, objectName string) (io.ReadCloser, error) @@ -60,6 +62,7 @@ type Service interface { // AI summarization operations GetAISummary(ctx context.Context, id string) (*AISummaryResponse, error) TriggerAISummarize(ctx context.Context, id string) (*AISummarizeResponse, error) + TriggerAIAnalyze(ctx context.Context, id string) (*AIAnalyzeResponse, error) TriggerAITranslate(ctx context.Context, id, language string) (*AITranslateResponse, error) GetAllAISummaries(ctx context.Context) ([]ai_summarizer.TenderSummaryItem, error) // ListTendersWithScrapedDocuments lists tenders that have documents in MinIO. When syncMongoMetadata is true, @@ -451,14 +454,13 @@ func (s *tenderService) Delete(ctx context.Context, id string) error { return nil } -// GetDocumentSummaries retrieves document summaries for a specific tender +// GetDocumentSummaries retrieves per-document summaries from the AI service MinIO storage. func (s *tenderService) GetDocumentSummaries(ctx context.Context, id string) ([]DocumentSummary, error) { if id == "" { return nil, fmt.Errorf("tender ID cannot be empty") } - // Get tender from repository - tender, err := s.repository.GetByID(ctx, id) + t, err := s.repository.GetByID(ctx, id) if err != nil { s.logger.Error("Failed to get tender for document summaries", map[string]interface{}{ "tender_id": id, @@ -466,22 +468,60 @@ func (s *tenderService) GetDocumentSummaries(ctx context.Context, id string) ([] }) return nil, fmt.Errorf("failed to get tender: %w", err) } - - if tender == nil { + if t == nil { return nil, fmt.Errorf("tender not found") } - - // Return document summaries (empty slice if none exist) - if tender.DocumentSummaries == nil { + if strings.TrimSpace(t.NoticePublicationID) == "" || strings.TrimSpace(t.ContractFolderID) == "" { + return []DocumentSummary{}, nil + } + if s.aiSummarizerStorage == nil { return []DocumentSummary{}, nil } + stored, err := s.aiSummarizerStorage.GetDocumentSummariesFromStorage(ctx, t.ContractFolderID, t.NoticePublicationID) + if errors.Is(err, ai_summarizer.ErrSummaryNotReady) { + return []DocumentSummary{}, nil + } + if errors.Is(err, ai_summarizer.ErrObjectNotFound) { + return []DocumentSummary{}, nil + } + if err != nil { + if errors.Is(err, ai_summarizer.ErrMinIOUnavailable) { + return nil, fmt.Errorf("failed to get document summaries: MinIO connection unavailable: %w", err) + } + s.logger.Error("Failed to get document summaries from storage", map[string]interface{}{ + "tender_id": id, + "error": err.Error(), + }) + return nil, fmt.Errorf("failed to get document summaries: %w", err) + } + + summaries := mapAIDocumentSummaries(stored) s.logger.Info("Retrieved document summaries for tender", map[string]interface{}{ "tender_id": id, - "summaries_count": len(tender.DocumentSummaries), + "summaries_count": len(summaries), + "source": "storage", }) - return tender.DocumentSummaries, nil + return summaries, nil +} + +func mapAIDocumentSummaries(items []ai_summarizer.DocumentSummary) []DocumentSummary { + if len(items) == 0 { + return []DocumentSummary{} + } + out := make([]DocumentSummary, 0, len(items)) + for _, item := range items { + out = append(out, DocumentSummary{ + DocumentName: item.DocumentName, + Summary: item.Summary, + SummaryLanguage: "en", + DocumentType: item.DocumentType, + SummaryModel: item.SummaryModel, + Error: item.Error, + }) + } + return out } // GetDocuments retrieves scraped document metadata for a tender. @@ -1175,23 +1215,7 @@ func (s *tenderService) TriggerAISummarize(ctx context.Context, id string) (*AIS return nil, fmt.Errorf("tender has no contract folder ID") } - authorityName := "" - if t.BuyerOrganization != nil { - authorityName = t.BuyerOrganization.Name - } - - reqBody := ai_summarizer.NewSummarizeRequest(ai_summarizer.SummarizeRequestInput{ - ContractFolderID: t.ContractFolderID, - NoticePublicationID: t.NoticePublicationID, - DocumentURL: t.DocumentURI, - Title: t.Title, - Description: t.Description, - Country: t.CountryCode, - Currency: t.Currency, - SubmissionDeadline: t.SubmissionDeadline, - EstimatedValue: t.EstimatedValue, - AuthorityName: authorityName, - }) + reqBody := ai_summarizer.NewSummarizeRequest(t.ContractFolderID, t.NoticePublicationID) s.logger.Info("Triggering on-demand AI summarization", map[string]interface{}{ "tender_id": id, @@ -1247,6 +1271,89 @@ func (s *tenderService) TriggerAISummarize(ctx context.Context, id string) (*AIS }, nil } +// TriggerAIAnalyze triggers on-demand agentic analysis for a tender. +func (s *tenderService) TriggerAIAnalyze(ctx context.Context, id string) (*AIAnalyzeResponse, error) { + if id == "" { + return nil, fmt.Errorf("tender ID cannot be empty") + } + if s.aiSummarizerClient == nil { + return nil, fmt.Errorf("AI summarizer service is not configured") + } + + t, err := s.repository.GetByID(ctx, id) + if err != nil { + s.logger.Error("Failed to get tender for on-demand analysis", map[string]interface{}{ + "tender_id": id, + "error": err.Error(), + }) + return nil, fmt.Errorf("failed to get tender: %w", err) + } + if t == nil { + return nil, fmt.Errorf("tender not found") + } + if t.NoticePublicationID == "" { + return nil, fmt.Errorf("tender has no notice publication ID") + } + if strings.TrimSpace(t.ContractFolderID) == "" { + return nil, fmt.Errorf("tender has no contract folder ID") + } + + reqBody := ai_summarizer.NewAnalyzeRequest(t.ContractFolderID, t.NoticePublicationID) + s.logger.Info("Triggering on-demand AI analysis", map[string]interface{}{ + "tender_id": id, + "contract_folder_id": t.ContractFolderID, + "notice_publication_id": t.NoticePublicationID, + }) + + resp, err := s.aiSummarizerClient.FetchAnalyzeOnDemand(ctx, reqBody) + if err != nil { + s.logger.Error("On-demand AI analysis failed", map[string]interface{}{ + "tender_id": id, + "contract_folder_id": t.ContractFolderID, + "notice_publication_id": t.NoticePublicationID, + "error": err.Error(), + }) + return nil, fmt.Errorf("AI analysis request failed: %w", err) + } + + documents := make([]AIAnalyzeDocument, 0, len(resp.Documents)) + for _, doc := range resp.Documents { + documents = append(documents, AIAnalyzeDocument{ + Filename: doc.Filename, + SizeBytes: doc.SizeBytes, + TimesRead: doc.TimesRead, + TimesQueried: doc.TimesQueried, + Description: doc.Description, + }) + } + + noticePublicationID := strings.TrimSpace(resp.NoticePublicationID) + if noticePublicationID == "" { + noticePublicationID = t.NoticePublicationID + } + contractFolderID := strings.TrimSpace(resp.ContractFolderID) + if contractFolderID == "" { + contractFolderID = t.ContractFolderID + } + + s.logger.Info("On-demand AI analysis completed", map[string]interface{}{ + "tender_id": id, + "contract_folder_id": contractFolderID, + "notice_publication_id": noticePublicationID, + "iterations": resp.Iterations, + }) + + return &AIAnalyzeResponse{ + ContractFolderID: contractFolderID, + NoticePublicationID: noticePublicationID, + Summary: resp.Summary, + Documents: documents, + ToolCallsLog: resp.ToolCallsLog, + Iterations: resp.Iterations, + Model: resp.Model, + }, nil +} + // GetAllAISummaries retrieves all AI-generated summaries from the MinIO storage. // It lists all tender.json files in the bucket and returns summaries for tenders // where the AI pipeline has completed processing. @@ -1510,11 +1617,12 @@ func (s *tenderService) buildAITranslateResponse(t *Tender, language, title, des func (s *tenderService) resolveTranslation(ctx context.Context, t *Tender, language, requestSource string) (string, string, bool, error) { if s.aiSummarizerStorage != nil { - stored, err := s.aiSummarizerStorage.GetTranslationFromStorage(ctx, t.ContractFolderID, t.NoticePublicationID, language) - if err == nil { + stored, _, err := s.lookupTranslationForTender(ctx, t, language) + if err == nil && strings.TrimSpace(stored.Title) != "" { return stored.Title, stored.Description, false, nil } - if !errors.Is(err, ai_summarizer.ErrTranslationNotReady) && + if err != nil && + !errors.Is(err, ai_summarizer.ErrTranslationNotReady) && !errors.Is(err, ai_summarizer.ErrObjectNotFound) && !errors.Is(err, ai_summarizer.ErrNoTranslation) { s.logger.Debug("Translation not available from storage", map[string]interface{}{ @@ -1528,15 +1636,26 @@ func (s *tenderService) resolveTranslation(ctx context.Context, t *Tender, langu resp, err := s.aiSummarizerClient.FetchTranslationOnDemand(ctx, ai_summarizer.TranslateRequest{ ContractFolderID: t.ContractFolderID, NoticePublicationID: t.NoticePublicationID, - Title: t.Title, - Description: t.Description, Language: language, RequestSource: requestSource, }) if err != nil { return "", "", false, err } - return resp.TranslatedTitle, resp.TranslatedDescription, true, nil + + title := strings.TrimSpace(resp.TranslatedTitle) + description := resp.TranslatedDescription + if s.aiSummarizerStorage != nil { + if stored, _, storageErr := s.lookupTranslationForTender(ctx, t, language); storageErr == nil && strings.TrimSpace(stored.Title) != "" { + title = stored.Title + description = stored.Description + } + } + if title == "" { + title = strings.TrimSpace(resp.TranslatedTitle) + } + + return title, description, true, nil } func (s *tenderService) pickResponseLanguage(language *string) string { diff --git a/pkg/ai_summarizer/client.go b/pkg/ai_summarizer/client.go index 27ed873..fc1a90d 100644 --- a/pkg/ai_summarizer/client.go +++ b/pkg/ai_summarizer/client.go @@ -169,6 +169,118 @@ func (c *Client) TriggerPipelineTranslate(ctx context.Context, languages []strin return &result, nil } +// FetchAnalyzeOnDemand calls POST /ai/analyze for one tender. +func (c *Client) FetchAnalyzeOnDemand(ctx context.Context, reqBody AnalyzeRequest) (*AnalyzeResponse, error) { + jsonBody, err := json.Marshal(reqBody) + if err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to marshal analyze request body: %w", err) + } + + url := c.config.APIBaseURL + "/ai/analyze" + httpResp, bodyBytes, err := c.doRawPost(ctx, url, jsonBody) + if err != nil { + return nil, err + } + defer httpResp.Body.Close() + + if httpResp.StatusCode == http.StatusNotFound { + return nil, fmt.Errorf("%w: status 404, body: %s", ErrObjectNotFound, string(bodyBytes)) + } + if httpResp.StatusCode >= 400 { + return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + } + + var result AnalyzeResponse + if err := json.Unmarshal(bodyBytes, &result); err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to decode analyze response JSON: %w", err) + } + + c.logger.Info("AI analyze request succeeded", map[string]interface{}{ + "contract_folder_id": result.ContractFolderID, + "notice_publication_id": result.NoticePublicationID, + "iterations": result.Iterations, + }) + + return &result, nil +} + +// TriggerPipelineSync calls POST /pipeline/sync. +func (c *Client) TriggerPipelineSync(ctx context.Context) (*PipelineActionResponse, error) { + return c.triggerPipelineAction(ctx, "/pipeline/sync", "pipeline sync") +} + +// TriggerPipelineScrape calls POST /pipeline/scrape. +func (c *Client) TriggerPipelineScrape(ctx context.Context) (*PipelineActionResponse, error) { + return c.triggerPipelineAction(ctx, "/pipeline/scrape", "pipeline scrape") +} + +// TriggerPipelineSummarize calls POST /pipeline/summarize. +func (c *Client) TriggerPipelineSummarize(ctx context.Context) (*PipelineActionResponse, error) { + return c.triggerPipelineAction(ctx, "/pipeline/summarize", "pipeline summarize") +} + +// TriggerPipelineAnalyze calls POST /pipeline/analyze. +func (c *Client) TriggerPipelineAnalyze(ctx context.Context) (*PipelineActionResponse, error) { + return c.triggerPipelineAction(ctx, "/pipeline/analyze", "pipeline analyze") +} + +// TriggerPipelineDailyRun calls POST /pipeline/daily-run. +func (c *Client) TriggerPipelineDailyRun(ctx context.Context) (*PipelineActionResponse, error) { + return c.triggerPipelineAction(ctx, "/pipeline/daily-run", "pipeline daily-run") +} + +func (c *Client) triggerPipelineAction(ctx context.Context, path, label string) (*PipelineActionResponse, error) { + url := c.config.APIBaseURL + path + httpResp, bodyBytes, err := c.doRawPost(ctx, url, nil) + if err != nil { + return nil, err + } + defer httpResp.Body.Close() + + if httpResp.StatusCode >= 400 { + return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + } + + var result PipelineActionResponse + if len(bodyBytes) > 0 { + if err := json.Unmarshal(bodyBytes, &result); err != nil { + return nil, fmt.Errorf("ai_summarizer: failed to decode %s response: %w", label, err) + } + } + + c.logger.Info("AI pipeline action triggered", map[string]interface{}{ + "path": path, + "status": result.Status, + }) + + return &result, nil +} + +func (c *Client) doRawPost(ctx context.Context, url string, jsonBody []byte) (*http.Response, []byte, error) { + var body io.Reader + if len(jsonBody) > 0 { + body = bytes.NewReader(jsonBody) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, body) + if err != nil { + return nil, nil, fmt.Errorf("ai_summarizer: failed to create request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + + httpResp, err := c.httpClient.Do(req) + if err != nil { + return nil, nil, fmt.Errorf("ai_summarizer: request error: %w", err) + } + + bodyBytes, err := io.ReadAll(httpResp.Body) + if err != nil { + httpResp.Body.Close() + return nil, nil, fmt.Errorf("ai_summarizer: failed to read response body: %w", err) + } + return httpResp, bodyBytes, nil +} + // doPost performs a single POST request and parses the response. func (c *Client) doPost(ctx context.Context, url string, jsonBody []byte) (*SummarizeResponse, error) { req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(jsonBody)) diff --git a/pkg/ai_summarizer/entities.go b/pkg/ai_summarizer/entities.go index 9b04ee0..430d04b 100644 --- a/pkg/ai_summarizer/entities.go +++ b/pkg/ai_summarizer/entities.go @@ -2,58 +2,20 @@ package ai_summarizer import ( "strings" - "time" ) -// SummarizeRequest represents the request payload sent to the AI service -// POST /ai/summarize endpoint for on-demand summarization. +// SummarizeRequest is the payload for POST /ai/summarize. type SummarizeRequest struct { - ContractFolderID string `json:"contract_folder_id"` - NoticePublicationID string `json:"notice_publication_id"` - DocumentURL string `json:"document_url,omitempty"` - Title string `json:"title,omitempty"` - Description string `json:"description,omitempty"` - Country string `json:"country,omitempty"` - EstimatedValue *float64 `json:"estimated_value,omitempty"` - Currency string `json:"currency,omitempty"` - SubmissionDeadline string `json:"submission_deadline,omitempty"` - AuthorityName string `json:"authority_name,omitempty"` -} - -// SummarizeRequestInput collects fields used to build SummarizeRequest. -type SummarizeRequestInput struct { - ContractFolderID string - NoticePublicationID string - DocumentURL string - Title string - Description string - Country string - Currency string - SubmissionDeadline int64 // Unix seconds; omitted from request when zero - EstimatedValue float64 - AuthorityName string + ContractFolderID string `json:"contract_folder_id"` + NoticePublicationID string `json:"notice_publication_id"` } // NewSummarizeRequest builds a SummarizeRequest for POST /ai/summarize. -func NewSummarizeRequest(in SummarizeRequestInput) SummarizeRequest { - req := SummarizeRequest{ - ContractFolderID: strings.TrimSpace(in.ContractFolderID), - NoticePublicationID: strings.TrimSpace(in.NoticePublicationID), - DocumentURL: strings.TrimSpace(in.DocumentURL), - Title: in.Title, - Description: in.Description, - Country: in.Country, - Currency: in.Currency, - AuthorityName: strings.TrimSpace(in.AuthorityName), +func NewSummarizeRequest(contractFolderID, noticePublicationID string) SummarizeRequest { + return SummarizeRequest{ + ContractFolderID: strings.TrimSpace(contractFolderID), + NoticePublicationID: strings.TrimSpace(noticePublicationID), } - if in.EstimatedValue > 0 { - v := in.EstimatedValue - req.EstimatedValue = &v - } - if in.SubmissionDeadline > 0 { - req.SubmissionDeadline = time.Unix(in.SubmissionDeadline, 0).UTC().Format(time.RFC3339) - } - return req } // SummarizeResponse represents the response from the AI summarization API. @@ -63,6 +25,8 @@ type SummarizeResponse struct { NoticeID string `json:"notice_id"` // legacy field name Summaries []DocumentSummary `json:"summaries"` OverallSummary *string `json:"overall_summary"` + RollupS float64 `json:"rollup_s"` + TotalS float64 `json:"total_s"` } // ResolvedNoticePublicationID returns the notice publication ID from the response. @@ -82,7 +46,7 @@ type DocumentSummary struct { DocumentType string `json:"document_type"` Summary string `json:"summary"` SummaryModel string `json:"summary_model"` - Error string `json:"error"` // empty string when no error + Error string `json:"error"` // empty when no error } // ProcedureDocumentsSummary reports scraped documents stored under one procedure folder in MinIO. @@ -101,13 +65,10 @@ type StoredDocument struct { DocumentType string `json:"document_type,omitempty"` } -// TranslateRequest represents the request payload sent to the AI service -// POST /ai/translate endpoint. +// TranslateRequest is the payload for POST /ai/translate. type TranslateRequest struct { ContractFolderID string `json:"contract_folder_id"` NoticePublicationID string `json:"notice_publication_id"` - Title string `json:"title"` - Description string `json:"description"` Language string `json:"language"` RequestSource string `json:"-"` // internal: daily_job | manual_trigger } @@ -126,6 +87,41 @@ type TranslateResponse struct { TranslatedDescription string `json:"translated_description"` } +// AnalyzeRequest is the payload for POST /ai/analyze. +type AnalyzeRequest struct { + ContractFolderID string `json:"contract_folder_id"` + NoticePublicationID string `json:"notice_publication_id"` +} + +// NewAnalyzeRequest builds an AnalyzeRequest for POST /ai/analyze. +func NewAnalyzeRequest(contractFolderID, noticePublicationID string) AnalyzeRequest { + return AnalyzeRequest{ + ContractFolderID: strings.TrimSpace(contractFolderID), + NoticePublicationID: strings.TrimSpace(noticePublicationID), + } +} + +// AnalyzeDocumentStats describes one document touched during agentic analysis. +type AnalyzeDocumentStats struct { + Filename string `json:"filename"` + SizeBytes int64 `json:"size_bytes"` + TimesRead int `json:"times_read"` + TimesQueried int `json:"times_queried"` + Description string `json:"description"` +} + +// AnalyzeResponse is the payload from POST /ai/analyze. +type AnalyzeResponse struct { + ContractFolderID string `json:"contract_folder_id"` + NoticePublicationID string `json:"notice_publication_id"` + Summary string `json:"summary"` + Documents []AnalyzeDocumentStats `json:"documents"` + Scratchpad map[string]interface{} `json:"scratchpad"` + ToolCallsLog []string `json:"tool_calls_log"` + Iterations int `json:"iterations"` + Model string `json:"model"` +} + // PipelineTranslateRequest represents the request payload for POST /pipeline/translate. type PipelineTranslateRequest struct { Languages []string `json:"languages"` @@ -137,6 +133,11 @@ type PipelineTranslateResponse struct { Languages []string `json:"languages"` } +// PipelineActionResponse is returned by fire-and-forget pipeline endpoints. +type PipelineActionResponse struct { + Status string `json:"status"` +} + // StoredTranslation is one language entry inside tender.json translations map. type StoredTranslation struct { Title string `json:"title"` @@ -173,6 +174,7 @@ type TenderJSON struct { Summary string `json:"summary"` OverallSummary *string `json:"overall_summary"` OverallSummaryCamel *string `json:"overallSummary"` + Summaries []DocumentSummary `json:"summaries"` SummarizeStatus SummarizeStatus `json:"summarize_status"` SummarizeStatusCamel SummarizeStatus `json:"summarizeStatus"` Translations map[string]StoredTranslation `json:"translations"` @@ -225,23 +227,31 @@ func (t *TenderJSON) translationsDone() []string { return t.TranslationStatusCamel.languagesDone() } -// translationForLanguage returns stored translation text when the pipeline marked -// the language done and non-empty title exists. +// storedTranslationText returns translation text from the translations map when +// title is non-empty. It does not require translation_status.done — on-demand +// writes may persist translations before updating status. +func (t *TenderJSON) storedTranslationText(language string) (StoredTranslation, bool) { + if t == nil || t.Translations == nil { + return StoredTranslation{}, false + } + lang := strings.ToLower(strings.TrimSpace(language)) + for key, entry := range t.Translations { + if strings.ToLower(strings.TrimSpace(key)) != lang { + continue + } + if strings.TrimSpace(entry.Title) == "" { + continue + } + return entry, true + } + return StoredTranslation{}, false +} + +// translationForLanguage returns stored translation when the pipeline marked the +// language done and non-empty title exists. func (t *TenderJSON) translationForLanguage(language string) (StoredTranslation, bool) { if t == nil || !languageInDone(t.translationsDone(), language) { return StoredTranslation{}, false } - lang := strings.ToLower(strings.TrimSpace(language)) - if t.Translations != nil { - for key, entry := range t.Translations { - if strings.ToLower(strings.TrimSpace(key)) != lang { - continue - } - if strings.TrimSpace(entry.Title) == "" { - continue - } - return entry, true - } - } - return StoredTranslation{}, false + return t.storedTranslationText(language) } diff --git a/pkg/ai_summarizer/storage.go b/pkg/ai_summarizer/storage.go index c0bfa5f..2dd75e4 100644 --- a/pkg/ai_summarizer/storage.go +++ b/pkg/ai_summarizer/storage.go @@ -326,8 +326,9 @@ func (s *StorageClient) GetSummaryFromStorage(ctx context.Context, contractFolde return s.getSummaryFromObjectKey(ctx, prefix+"tender.json", contractFolderID, noticePublicationID) } -// GetTranslationFromStorage reads tender.json and returns the translation for a language -// when translation_status.done includes that language. +// GetTranslationFromStorage reads tender.json and returns the translation for a language. +// It prefers entries marked done in translation_status; if missing, it falls back to +// translations[] when title is non-empty (on-demand API may lag status updates). func (s *StorageClient) GetTranslationFromStorage(ctx context.Context, contractFolderID, noticePublicationID, language string) (StoredTranslation, error) { if strings.TrimSpace(contractFolderID) == "" { return StoredTranslation{}, fmt.Errorf("ai_summarizer: contractFolderID is required") @@ -353,14 +354,16 @@ func (s *StorageClient) getTranslationFromObjectKey(ctx context.Context, objectK return StoredTranslation{}, err } + if entry, ok := tenderJSON.translationForLanguage(language); ok { + return entry, nil + } + if entry, ok := tenderJSON.storedTranslationText(language); ok { + return entry, nil + } if !languageInDone(tenderJSON.translationsDone(), language) { return StoredTranslation{}, ErrTranslationNotReady } - entry, ok := tenderJSON.translationForLanguage(language) - if !ok { - return StoredTranslation{}, ErrNoTranslation - } - return entry, nil + return StoredTranslation{}, ErrNoTranslation } // getSummaryFromObjectKey reads tender.json at objectKey and returns summary text when ready. @@ -432,6 +435,39 @@ func (s *StorageClient) getSummaryFromObjectKey(ctx context.Context, objectKey, return summaryText, nil } +// GetDocumentSummariesFromStorage reads per-document summaries from tender.json +// when summarize_status.done is true. +func (s *StorageClient) GetDocumentSummariesFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) ([]DocumentSummary, error) { + if strings.TrimSpace(contractFolderID) == "" { + return nil, fmt.Errorf("ai_summarizer: contractFolderID is required") + } + if strings.TrimSpace(noticePublicationID) == "" { + return nil, fmt.Errorf("ai_summarizer: noticePublicationID is required") + } + + prefix, err := s.resolveNoticeStoragePrefix(ctx, contractFolderID, noticePublicationID) + if err != nil { + return nil, err + } + + tenderJSON, err := s.readTenderJSONAtKey(ctx, prefix+"tender.json") + if err != nil { + return nil, err + } + + done, _ := tenderJSON.resolved() + if !done { + return nil, ErrSummaryNotReady + } + if len(tenderJSON.Summaries) == 0 { + return []DocumentSummary{}, nil + } + + out := make([]DocumentSummary, len(tenderJSON.Summaries)) + copy(out, tenderJSON.Summaries) + return out, nil +} + // IsSummaryReady is a convenience method that checks whether the tender.json // exists and the summarize_status.done flag is true, without returning the // actual summary text. From d07e1c9cf07da405be13ebd3e9001f02d0a23c84 Mon Sep 17 00:00:00 2001 From: Mazyar Date: Mon, 8 Jun 2026 20:24:02 +0330 Subject: [PATCH 2/3] Fix summarizer worker infinite loop and map AI trigger errors to safe HTTP responses. Paginate un-summarized tenders and mark rows missing notice/folder IDs as handled so the worker cannot spin forever. Return 400/404 for validation and not-found cases on AI summarize/analyze triggers instead of leaking internal errors as 500. --- cmd/worker/workers/summarizer.go | 11 ++++++++++- internal/tender/errors.go | 13 +++++++++++++ internal/tender/handler.go | 20 +++++++++++++++++-- internal/tender/service.go | 33 ++++++++++++++++++++++---------- 4 files changed, 64 insertions(+), 13 deletions(-) create mode 100644 internal/tender/errors.go diff --git a/cmd/worker/workers/summarizer.go b/cmd/worker/workers/summarizer.go index 51f80f7..dcc352f 100644 --- a/cmd/worker/workers/summarizer.go +++ b/cmd/worker/workers/summarizer.go @@ -54,8 +54,9 @@ func (w *DocumentSummarizationWorker) Run() { } limit := 5 + skip := 0 for { - tenders, totalCount, err := w.TenderRepo.GetUnSummarizedTenders(context.Background(), limit, 0) + tenders, totalCount, err := w.TenderRepo.GetUnSummarizedTenders(context.Background(), limit, skip) if err != nil { w.Logger.Error("Failed to get un-summarized tenders", map[string]interface{}{ "error": err.Error(), @@ -66,6 +67,7 @@ func (w *DocumentSummarizationWorker) Run() { w.Logger.Info("Found tenders for document summarization", map[string]interface{}{ "count": len(tenders), "total_count": totalCount, + "skip": skip, }) if len(tenders) == 0 { @@ -80,6 +82,11 @@ func (w *DocumentSummarizationWorker) Run() { }) w.summarizeTender(t) } + + skip += len(tenders) + if int64(skip) >= totalCount { + break + } } w.Logger.Info("Document summarization worker completed", map[string]interface{}{}) @@ -90,12 +97,14 @@ func (w *DocumentSummarizationWorker) summarizeTender(t *tender.Tender) { w.Logger.Warn("Skipping tender summarization: missing notice_publication_id", map[string]interface{}{ "tender_id": t.ID.Hex(), }) + w.markTenderSummarized(t) return } if strings.TrimSpace(t.ContractFolderID) == "" { w.Logger.Warn("Skipping tender summarization: missing contract_folder_id", map[string]interface{}{ "tender_id": t.ID.Hex(), }) + w.markTenderSummarized(t) return } diff --git a/internal/tender/errors.go b/internal/tender/errors.go new file mode 100644 index 0000000..a3f3486 --- /dev/null +++ b/internal/tender/errors.go @@ -0,0 +1,13 @@ +package tender + +import "errors" + +// Sentinel errors for AI trigger operations and HTTP mapping. +var ( + ErrEmptyTenderID = errors.New("tender: id is required") + ErrInvalidTenderID = errors.New("tender: invalid id") + ErrTenderNotFound = errors.New("tender: not found") + ErrTenderMissingNoticeID = errors.New("tender: missing notice publication id") + ErrTenderMissingContractFolderID = errors.New("tender: missing contract folder id") + ErrAINotConfigured = errors.New("tender: ai service not configured") +) diff --git a/internal/tender/handler.go b/internal/tender/handler.go index c239f1f..dc092c3 100644 --- a/internal/tender/handler.go +++ b/internal/tender/handler.go @@ -589,7 +589,7 @@ func (h *TenderHandler) TriggerAISummarize(c echo.Context) error { "tender_id": id, "error": err.Error(), }) - return response.InternalServerError(c, err.Error()) + return mapAITriggerHTTPError(c, err, "AI summarization") } return response.Success(c, result, "AI summarization completed successfully") @@ -619,12 +619,28 @@ func (h *TenderHandler) TriggerAIAnalyze(c echo.Context) error { "tender_id": id, "error": err.Error(), }) - return response.InternalServerError(c, err.Error()) + return mapAITriggerHTTPError(c, err, "AI analysis") } return response.Success(c, result, "AI analysis completed successfully") } +// mapAITriggerHTTPError maps AI trigger service errors to safe HTTP responses. +func mapAITriggerHTTPError(c echo.Context, err error, action string) error { + switch { + case errors.Is(err, ErrEmptyTenderID), errors.Is(err, ErrInvalidTenderID): + return response.BadRequest(c, "Invalid request", "Invalid tender ID") + case errors.Is(err, ErrTenderNotFound), errors.Is(err, orm.ErrDocumentNotFound): + return response.NotFound(c, "Tender not found") + case errors.Is(err, ErrTenderMissingNoticeID), errors.Is(err, ErrTenderMissingContractFolderID): + return response.BadRequest(c, "Tender not eligible", "Tender is missing required identifiers for AI processing") + case errors.Is(err, ErrAINotConfigured): + return response.InternalServerError(c, "AI service is not available") + default: + return response.InternalServerError(c, action+" failed") + } +} + // TriggerAITranslate triggers on-demand AI translation for a tender // @Summary Trigger AI translation // @Description Trigger on-demand AI translation for tender title and description (cached in MinIO by the AI service) diff --git a/internal/tender/service.go b/internal/tender/service.go index 44cd986..840c9e2 100644 --- a/internal/tender/service.go +++ b/internal/tender/service.go @@ -15,6 +15,7 @@ import ( "tm/internal/customer" "tm/pkg/ai_summarizer" "tm/pkg/logger" + orm "tm/pkg/mongo" "tm/pkg/response" ) @@ -1187,11 +1188,11 @@ func (s *tenderService) GetAISummary(ctx context.Context, id string) (*AISummary // It calls the external AI service HTTP API to generate summaries. func (s *tenderService) TriggerAISummarize(ctx context.Context, id string) (*AISummarizeResponse, error) { if id == "" { - return nil, fmt.Errorf("tender ID cannot be empty") + return nil, ErrEmptyTenderID } if s.aiSummarizerClient == nil { - return nil, fmt.Errorf("AI summarizer service is not configured") + return nil, ErrAINotConfigured } // Get tender to build the request @@ -1201,18 +1202,24 @@ func (s *tenderService) TriggerAISummarize(ctx context.Context, id string) (*AIS "tender_id": id, "error": err.Error(), }) + if errors.Is(err, orm.ErrDocumentNotFound) { + return nil, ErrTenderNotFound + } + if strings.Contains(err.Error(), "invalid id format") { + return nil, ErrInvalidTenderID + } return nil, fmt.Errorf("failed to get tender: %w", err) } if t == nil { - return nil, fmt.Errorf("tender not found") + return nil, ErrTenderNotFound } if t.NoticePublicationID == "" { - return nil, fmt.Errorf("tender has no notice publication ID") + return nil, ErrTenderMissingNoticeID } if strings.TrimSpace(t.ContractFolderID) == "" { - return nil, fmt.Errorf("tender has no contract folder ID") + return nil, ErrTenderMissingContractFolderID } reqBody := ai_summarizer.NewSummarizeRequest(t.ContractFolderID, t.NoticePublicationID) @@ -1274,10 +1281,10 @@ func (s *tenderService) TriggerAISummarize(ctx context.Context, id string) (*AIS // TriggerAIAnalyze triggers on-demand agentic analysis for a tender. func (s *tenderService) TriggerAIAnalyze(ctx context.Context, id string) (*AIAnalyzeResponse, error) { if id == "" { - return nil, fmt.Errorf("tender ID cannot be empty") + return nil, ErrEmptyTenderID } if s.aiSummarizerClient == nil { - return nil, fmt.Errorf("AI summarizer service is not configured") + return nil, ErrAINotConfigured } t, err := s.repository.GetByID(ctx, id) @@ -1286,16 +1293,22 @@ func (s *tenderService) TriggerAIAnalyze(ctx context.Context, id string) (*AIAna "tender_id": id, "error": err.Error(), }) + if errors.Is(err, orm.ErrDocumentNotFound) { + return nil, ErrTenderNotFound + } + if strings.Contains(err.Error(), "invalid id format") { + return nil, ErrInvalidTenderID + } return nil, fmt.Errorf("failed to get tender: %w", err) } if t == nil { - return nil, fmt.Errorf("tender not found") + return nil, ErrTenderNotFound } if t.NoticePublicationID == "" { - return nil, fmt.Errorf("tender has no notice publication ID") + return nil, ErrTenderMissingNoticeID } if strings.TrimSpace(t.ContractFolderID) == "" { - return nil, fmt.Errorf("tender has no contract folder ID") + return nil, ErrTenderMissingContractFolderID } reqBody := ai_summarizer.NewAnalyzeRequest(t.ContractFolderID, t.NoticePublicationID) From 9cd22c24b6ac0949554b24efb794608322373cc3 Mon Sep 17 00:00:00 2001 From: Mazyar Date: Mon, 8 Jun 2026 23:49:51 +0330 Subject: [PATCH 3/3] Refactor queue management by removing deprecated components and updating configuration - Removed the QueueConfig structure and related queue management files as they are no longer in use. - Updated the worker initialization to reflect the removal of queue-related configurations. - Cleaned up the bootstrap process by eliminating deprecated logging related to the queue system. This update streamlines the worker's configuration and prepares the codebase for future enhancements without the legacy queue management components. --- cmd/worker/bootstrap/bootstrap.go | 8 - cmd/worker/bootstrap/config.go | 16 -- cmd/worker/workers/queue_producer.go | 165 ------------- internal/tender/repository.go | 31 --- pkg/ai_summarizer/client.go | 20 -- pkg/queue/config.go | 53 ---- pkg/queue/models.go | 85 ------- pkg/queue/queue.go | 357 --------------------------- 8 files changed, 735 deletions(-) delete mode 100644 cmd/worker/workers/queue_producer.go delete mode 100644 pkg/queue/config.go delete mode 100644 pkg/queue/models.go delete mode 100644 pkg/queue/queue.go diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index 21012c5..df337c3 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -85,8 +85,6 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger "notice_processing_limit": config.Worker.NoticeProcessingLimit, "notice_batch_pause": config.Worker.NoticeBatchPause.String(), "notice_fetch_error_backoff": config.Worker.NoticeFetchErrorBackoff.String(), - "queue_enabled": config.Queue.Enabled, - "queue_mode": config.Queue.Mode, }) // Initialize repositories noticeRepo := notice.NewRepository(mongoManager, appLogger) @@ -95,12 +93,6 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger // Create a single shared cron scheduler for all recurring jobs scheduler := schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo) - if config.Queue.Enabled { - appLogger.Warn("Redis document-scraper queue is deprecated; document scraping is owned by the AI service pipeline", map[string]interface{}{ - "queue_mode": config.Queue.Mode, - }) - } - // Document summarization via external AI service (requires AI_SUMMARIZER_API_BASE_URL) if aiClient != nil { scheduler.AddJob(schedule.Job{ diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index 874631e..47acc1b 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -15,25 +15,9 @@ type Config struct { Scraper config.ScraperConfig MinIO config.MinIOConfig AISummarizer AISummarizerConfig - Queue QueueConfig `env:"QUEUE_" envPrefix:"QUEUE_"` AlertMail string `env:"ALERT_MAIL" envDefault:"alerts@opplens.com"` } -// QueueConfig defines the queue configuration -type QueueConfig struct { - Enabled bool `env:"QUEUE_ENABLED" envDefault:"true"` - RedisURL string `env:"QUEUE_REDIS_URL" envDefault:"redis://localhost:6379"` - Mode string `env:"QUEUE_MODE" envDefault:"consumer"` // "producer", "consumer", or "both" - MaxRetries int `env:"QUEUE_MAX_RETRIES" envDefault:"3"` - RetryBackoff time.Duration `env:"QUEUE_RETRY_BACKOFF" envDefault:"10s"` - JobTimeout time.Duration `env:"QUEUE_JOB_TIMEOUT" envDefault:"5m"` - PoolSize int `env:"QUEUE_POOL_SIZE" envDefault:"10"` - IsDLQEnabled bool `env:"QUEUE_DLQ_ENABLED" envDefault:"true"` - Concurrency int `env:"QUEUE_CONCURRENCY" envDefault:"5"` - ProducerBatch int `env:"QUEUE_PRODUCER_BATCH" envDefault:"100"` - ProducerInterval string `env:"QUEUE_PRODUCER_INTERVAL" envDefault:"0 */6 * * * *"` // Every 6 hours -} - type WorkerConfig struct { Interval string `env:"WORKER_INTERVAL" envDefault:"* 10 * * * *"` NoticeProcessingLimit int `env:"WORKER_NOTICE_PROCESSING_LIMIT" envDefault:"0"` diff --git a/cmd/worker/workers/queue_producer.go b/cmd/worker/workers/queue_producer.go deleted file mode 100644 index 1927860..0000000 --- a/cmd/worker/workers/queue_producer.go +++ /dev/null @@ -1,165 +0,0 @@ -package workers - -import ( - "context" - "time" - "tm/internal/tender" - "tm/pkg/logger" - "tm/pkg/mongo" - "tm/pkg/queue" -) - -// QueueProducerWorker handles enqueuing unscraped tenders to the job queue -type QueueProducerWorker struct { - Mongo *mongo.ConnectionManager - Logger logger.Logger - TenderRepo tender.TenderRepository - Queue queue.Queue - BatchSize int -} - -// NewQueueProducerWorker creates a new queue producer worker -func NewQueueProducerWorker( - mongo *mongo.ConnectionManager, - logger logger.Logger, - tenderRepo tender.TenderRepository, - queue queue.Queue, - batchSize int, -) *QueueProducerWorker { - if batchSize <= 0 { - batchSize = 100 // Default batch size - } - - logger.Info("Creating new queue producer worker", map[string]interface{}{ - "batch_size": batchSize, - }) - - return &QueueProducerWorker{ - Mongo: mongo, - Logger: logger, - TenderRepo: tenderRepo, - Queue: queue, - BatchSize: batchSize, - } -} - -// Run continuously processes and enqueues unscraped tenders -func (w *QueueProducerWorker) Run() { - w.Logger.Info("Queue producer worker started", map[string]interface{}{}) - - defer func() { - w.Logger.Info("Queue producer worker stopped", map[string]interface{}{}) - }() - - batchIndex := 0 - hasMore := true - - for hasMore { - w.Logger.Info("Queue producer fetching batch of tenders", map[string]interface{}{ - "batch_index": batchIndex, - "batch_size": w.BatchSize, - }) - - // Get batch of unscraped tenders - tenders, totalCount, err := w.TenderRepo.GetUnScrapedTenders( - context.Background(), - w.BatchSize, - batchIndex*w.BatchSize, - ) - - if err != nil { - w.Logger.Error("Failed to fetch unscraped tenders", map[string]interface{}{ - "batch_index": batchIndex, - "error": err.Error(), - }) - // Wait before retrying - time.Sleep(30 * time.Second) - continue - } - - w.Logger.Info("Fetched tender batch", map[string]interface{}{ - "batch_size": len(tenders), - "total_count": totalCount, - "batch_index": batchIndex, - }) - - if len(tenders) == 0 { - // No more tenders in this batch - hasMore = false - w.Logger.Info("No more unscraped tenders found", map[string]interface{}{ - "total_processed": totalCount, - }) - break - } - - // Enqueue each tender - successCount := 0 - failCount := 0 - - for _, t := range tenders { - if err := w.enqueueTender(&t); err != nil { - w.Logger.Error("Failed to enqueue tender", map[string]interface{}{ - "tender_id": t.ID.Hex(), - "notice_publication_id": t.NoticePublicationID, - "error": err.Error(), - }) - failCount++ - } else { - successCount++ - } - } - - w.Logger.Info("Batch enqueuing completed", map[string]interface{}{ - "success_count": successCount, - "fail_count": failCount, - "batch_index": batchIndex, - }) - - // Move to next batch - batchIndex++ - - // Check if there are more tenders - if int64(batchIndex*w.BatchSize) >= totalCount { - hasMore = false - } - - // Avoid tight loop - brief pause between batches - time.Sleep(1 * time.Second) - } -} - -// enqueueTender adds a single tender to the queue -func (w *QueueProducerWorker) enqueueTender(t *tender.Tender) error { - // Skip if missing required fields - if t.NoticePublicationID == "" || t.DocumentURI == "" { - w.Logger.Warn("Skipping tender - missing required fields", map[string]interface{}{ - "tender_id": t.ID.Hex(), - "notice_publication_id": t.NoticePublicationID, - "document_uri": t.DocumentURI, - }) - return nil - } - - // Create job - job := queue.NewDocumentScraperJob( - t.NoticePublicationID, - t.DocumentURI, - t.ID.Hex(), - queue.DefaultMaxRetries, - ) - - // Enqueue job - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - if err := w.Queue.Enqueue(ctx, job); err != nil { - return err - } - - return nil -} - -// RunOnce processes all unscraped tenders in a single run (for scheduled use) -func (w *QueueProducerWorker) RunOnce() { - w.Run() -} diff --git a/internal/tender/repository.go b/internal/tender/repository.go index 3b03226..6e9f57a 100644 --- a/internal/tender/repository.go +++ b/internal/tender/repository.go @@ -31,7 +31,6 @@ type TenderRepository interface { GetTenderCountByClassification(ctx context.Context) (map[string]int64, error) Search(ctx context.Context, form *SearchForm, pagination *response.Pagination) (*SearchPageResult, error) GetExpiredTenders(ctx context.Context, beforeDate int64) ([]Tender, error) - GetUnScrapedTenders(ctx context.Context, limit, skip int) ([]Tender, int64, error) GetPendingDocumentScrapeTenders(ctx context.Context, countryCodes []string, minDeadline int64, limit, skip int) ([]Tender, bool, error) GetUnSummarizedTenders(ctx context.Context, limit, skip int) ([]Tender, int64, error) GetUnTranslatedTenders(ctx context.Context, language string, limit, skip int) ([]Tender, int64, error) @@ -549,36 +548,6 @@ func (r *tenderRepository) GetExpiredTenders(ctx context.Context, beforeDate int return result.Items, nil } -// GetUnScrapedTenders retrieves tenders that haven't been processed for document scraping -func (r *tenderRepository) GetUnScrapedTenders(ctx context.Context, limit, skip int) ([]Tender, int64, error) { - filter := bson.M{ - "$or": []bson.M{ - {"processing_metadata.documents_scraped": bson.M{"$exists": false}}, - {"processing_metadata.documents_scraped": false}, - {"processing_metadata.documents_scraped": nil}, - }, - } - - pagination := orm.Pagination{ - Limit: limit, - Skip: skip, - SortField: "created_at", - SortOrder: 1, // Oldest first - } - - result, err := r.ormRepo.FindAll(ctx, filter, pagination) - if err != nil { - r.logger.Error("Failed to get un-scraped tenders", map[string]interface{}{ - "limit": limit, - "skip": skip, - "error": err.Error(), - }) - return nil, 0, err - } - - return result.Items, result.TotalCount, nil -} - // GetPendingDocumentScrapeTenders returns unscraped tenders for the given countries with an active deadline. func (r *tenderRepository) GetPendingDocumentScrapeTenders(ctx context.Context, countryCodes []string, minDeadline int64, limit, skip int) ([]Tender, bool, error) { if len(countryCodes) == 0 { diff --git a/pkg/ai_summarizer/client.go b/pkg/ai_summarizer/client.go index fc1a90d..70c3551 100644 --- a/pkg/ai_summarizer/client.go +++ b/pkg/ai_summarizer/client.go @@ -204,31 +204,11 @@ func (c *Client) FetchAnalyzeOnDemand(ctx context.Context, reqBody AnalyzeReques return &result, nil } -// TriggerPipelineSync calls POST /pipeline/sync. -func (c *Client) TriggerPipelineSync(ctx context.Context) (*PipelineActionResponse, error) { - return c.triggerPipelineAction(ctx, "/pipeline/sync", "pipeline sync") -} - -// TriggerPipelineScrape calls POST /pipeline/scrape. -func (c *Client) TriggerPipelineScrape(ctx context.Context) (*PipelineActionResponse, error) { - return c.triggerPipelineAction(ctx, "/pipeline/scrape", "pipeline scrape") -} - // TriggerPipelineSummarize calls POST /pipeline/summarize. func (c *Client) TriggerPipelineSummarize(ctx context.Context) (*PipelineActionResponse, error) { return c.triggerPipelineAction(ctx, "/pipeline/summarize", "pipeline summarize") } -// TriggerPipelineAnalyze calls POST /pipeline/analyze. -func (c *Client) TriggerPipelineAnalyze(ctx context.Context) (*PipelineActionResponse, error) { - return c.triggerPipelineAction(ctx, "/pipeline/analyze", "pipeline analyze") -} - -// TriggerPipelineDailyRun calls POST /pipeline/daily-run. -func (c *Client) TriggerPipelineDailyRun(ctx context.Context) (*PipelineActionResponse, error) { - return c.triggerPipelineAction(ctx, "/pipeline/daily-run", "pipeline daily-run") -} - func (c *Client) triggerPipelineAction(ctx context.Context, path, label string) (*PipelineActionResponse, error) { url := c.config.APIBaseURL + path httpResp, bodyBytes, err := c.doRawPost(ctx, url, nil) diff --git a/pkg/queue/config.go b/pkg/queue/config.go deleted file mode 100644 index 43c26b8..0000000 --- a/pkg/queue/config.go +++ /dev/null @@ -1,53 +0,0 @@ -package queue - -import ( - "time" -) - -/* -Queue Configuration and Constants - -This package provides a Redis-based job queue system for managing -document scraping tasks in a distributed manner. -*/ - -const ( - // Queue names - DocumentScraperQueue = "scraper:document:queue" - DocumentScraperDLQ = "scraper:document:dlq" // Dead Letter Queue - - // Job status keys - JobStatusPrefix = "job:" - QueueMetricsKey = "queue:metrics" - - // Default timeouts - DefaultJobTimeout = 5 * time.Minute - DefaultMaxRetries = 3 - DefaultRetryBackoff = 10 * time.Second - MaxRetryBackoff = 5 * time.Minute // Cap retry backoff at 5 minutes - InProgressExpiration = 10 * time.Minute // Job must complete within this time -) - -// QueueConfig represents configuration for the queue service -type QueueConfig struct { - RedisURL string - MaxRetries int - RetryBackoff time.Duration - JobTimeout time.Duration - PoolSize int - IsDLQEnabled bool - MaxQueueLength int // 0 means unlimited -} - -// DefaultConfig returns default queue configuration -func DefaultConfig() QueueConfig { - return QueueConfig{ - RedisURL: "redis://localhost:6379", - MaxRetries: DefaultMaxRetries, - RetryBackoff: DefaultRetryBackoff, - JobTimeout: DefaultJobTimeout, - PoolSize: 10, - IsDLQEnabled: true, - MaxQueueLength: 0, // Unlimited by default - } -} diff --git a/pkg/queue/models.go b/pkg/queue/models.go deleted file mode 100644 index 7568533..0000000 --- a/pkg/queue/models.go +++ /dev/null @@ -1,85 +0,0 @@ -package queue - -import ( - "encoding/json" - "time" -) - -// JobStatus represents the status of a queue job -type JobStatus string - -const ( - StatusPending JobStatus = "pending" - StatusInProgress JobStatus = "in_progress" - StatusCompleted JobStatus = "completed" - StatusFailed JobStatus = "failed" - StatusRetrying JobStatus = "retrying" -) - -// DocumentScraperJob represents a document scraping task in the queue -type DocumentScraperJob struct { - ID string `json:"id"` // Unique job ID (typically tender ID) - NoticeID string `json:"notice_id"` // Notice publication ID - NoticeURL string `json:"notice_url"` // Document URL to scrape - TenderID string `json:"tender_id,omitempty"` // Associated tender ID from MongoDB - CreatedAt int64 `json:"created_at"` - UpdatedAt int64 `json:"updated_at"` - Attempts int `json:"attempts"` - MaxRetries int `json:"max_retries"` - Status string `json:"status"` - Error string `json:"error,omitempty"` - NextRetryTime int64 `json:"next_retry_time,omitempty"` - CompletedAt int64 `json:"completed_at,omitempty"` - ProcessingStartedAt int64 `json:"processing_started_at,omitempty"` -} - -// NewDocumentScraperJob creates a new document scraper job -func NewDocumentScraperJob(noticeID, noticeURL, tenderID string, maxRetries int) *DocumentScraperJob { - now := time.Now().Unix() - return &DocumentScraperJob{ - ID: noticeID, - NoticeID: noticeID, - NoticeURL: noticeURL, - TenderID: tenderID, - CreatedAt: now, - UpdatedAt: now, - Attempts: 0, - MaxRetries: maxRetries, - Status: string(StatusPending), - } -} - -// ToJSON marshals the job to JSON -func (j *DocumentScraperJob) ToJSON() (string, error) { - data, err := json.Marshal(j) - return string(data), err -} - -// FromJSON unmarshals JSON to DocumentScraperJob -func (j *DocumentScraperJob) FromJSON(data string) error { - return json.Unmarshal([]byte(data), j) -} - -// QueueMetrics represents queue performance metrics -type QueueMetrics struct { - TotalJobs int64 `json:"total_jobs"` - PendingJobs int64 `json:"pending_jobs"` - InProgressJobs int64 `json:"in_progress_jobs"` - CompletedJobs int64 `json:"completed_jobs"` - FailedJobs int64 `json:"failed_jobs"` - AverageWaitTime time.Duration `json:"average_wait_time"` - AverageProcessTime time.Duration `json:"average_process_time"` - UpdatedAt int64 `json:"updated_at"` -} - -// JobResult represents the result of a completed job -type JobResult struct { - JobID string `json:"job_id"` - Status string `json:"status"` - Success bool `json:"success"` - Error string `json:"error,omitempty"` - CompletedAt int64 `json:"completed_at"` - Duration int64 `json:"duration_ms"` - UploadedCount int `json:"uploaded_count"` - Details map[string]interface{} `json:"details,omitempty"` -} diff --git a/pkg/queue/queue.go b/pkg/queue/queue.go deleted file mode 100644 index 4a9ba53..0000000 --- a/pkg/queue/queue.go +++ /dev/null @@ -1,357 +0,0 @@ -package queue - -import ( - "context" - "fmt" - "time" - "tm/pkg/logger" - - "github.com/redis/go-redis/v9" -) - -// Queue interface defines the queue operations -type Queue interface { - // Job operations - Enqueue(ctx context.Context, job *DocumentScraperJob) error - Dequeue(ctx context.Context) (*DocumentScraperJob, error) - AckJob(ctx context.Context, jobID string) error - NackJob(ctx context.Context, jobID string, err error) error - GetJobStatus(ctx context.Context, jobID string) (JobStatus, error) - - // Queue inspection - GetMetrics(ctx context.Context) (*QueueMetrics, error) - GetQueueSize(ctx context.Context) (int64, error) - GetDLQSize(ctx context.Context) (int64, error) - GetInProgressJobs(ctx context.Context) ([]DocumentScraperJob, error) - - // Cleanup operations - PurgeQueue(ctx context.Context) error - PurgeDLQ(ctx context.Context) error - Close() error -} - -// RedisQueue implements Queue interface using Redis -type RedisQueue struct { - client *redis.Client - config QueueConfig - logger logger.Logger -} - -// NewRedisQueue creates a new Redis-based queue -func NewRedisQueue(config QueueConfig, logger logger.Logger) (Queue, error) { - opt, err := redis.ParseURL(config.RedisURL) - if err != nil { - logger.Error("Failed to parse Redis URL", map[string]interface{}{ - "url": config.RedisURL, - "error": err.Error(), - }) - return nil, fmt.Errorf("failed to parse redis url: %w", err) - } - - opt.PoolSize = config.PoolSize - client := redis.NewClient(opt) - - // Test connection - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - if err := client.Ping(ctx).Err(); err != nil { - logger.Error("Failed to connect to Redis", map[string]interface{}{ - "error": err.Error(), - }) - return nil, fmt.Errorf("failed to connect to redis: %w", err) - } - - logger.Info("Redis queue initialized successfully", map[string]interface{}{ - "pool_size": config.PoolSize, - }) - - return &RedisQueue{ - client: client, - config: config, - logger: logger, - }, nil -} - -// Enqueue adds a job to the queue -func (q *RedisQueue) Enqueue(ctx context.Context, job *DocumentScraperJob) error { - if q.config.MaxQueueLength > 0 { - size, err := q.GetQueueSize(ctx) - if err != nil { - q.logger.Warn("Failed to check queue size", map[string]interface{}{ - "error": err.Error(), - }) - } else if size >= int64(q.config.MaxQueueLength) { - return fmt.Errorf("queue is full: %d/%d", size, q.config.MaxQueueLength) - } - } - - job.CreatedAt = time.Now().Unix() - job.UpdatedAt = time.Now().Unix() - job.Status = string(StatusPending) - - jsonData, err := job.ToJSON() - if err != nil { - q.logger.Error("Failed to marshal job", map[string]interface{}{ - "job_id": job.ID, - "error": err.Error(), - }) - return err - } - - // Push to queue - if err := q.client.RPush(ctx, DocumentScraperQueue, jsonData).Err(); err != nil { - q.logger.Error("Failed to enqueue job", map[string]interface{}{ - "job_id": job.ID, - "error": err.Error(), - }) - return err - } - - // Store job metadata with TTL - jobKey := fmt.Sprintf("%s%s", JobStatusPrefix, job.ID) - if err := q.client.Set(ctx, jobKey, string(StatusPending), 24*time.Hour).Err(); err != nil { - q.logger.Warn("Failed to set job status", map[string]interface{}{ - "job_id": job.ID, - "error": err.Error(), - }) - } - - q.logger.Info("Job enqueued successfully", map[string]interface{}{ - "job_id": job.ID, - "notice_id": job.NoticeID, - }) - - return nil -} - -// Dequeue retrieves and locks the next job from the queue -func (q *RedisQueue) Dequeue(ctx context.Context) (*DocumentScraperJob, error) { - // Use BLPOP to get next job with timeout - result, err := q.client.BLPop(ctx, 30*time.Second, DocumentScraperQueue).Result() - if err != nil { - if err == redis.Nil { - return nil, nil // Queue is empty - } - q.logger.Error("Failed to dequeue job", map[string]interface{}{ - "error": err.Error(), - }) - return nil, err - } - - if len(result) < 2 { - return nil, fmt.Errorf("invalid dequeue result") - } - - jobData := result[1] - - // Parse job - var job DocumentScraperJob - if err := job.FromJSON(jobData); err != nil { - q.logger.Error("Failed to parse job", map[string]interface{}{ - "error": err.Error(), - "data": jobData, - }) - return nil, err - } - - // Mark as in progress with TTL - job.Status = string(StatusInProgress) - job.Attempts++ - job.ProcessingStartedAt = time.Now().Unix() - job.UpdatedAt = time.Now().Unix() - - jobKey := fmt.Sprintf("%s%s", JobStatusPrefix, job.ID) - if err := q.client.Set(ctx, jobKey, string(StatusInProgress), InProgressExpiration).Err(); err != nil { - q.logger.Warn("Failed to update job status", map[string]interface{}{ - "job_id": job.ID, - "error": err.Error(), - }) - } - - q.logger.Info("Job dequeued successfully", map[string]interface{}{ - "job_id": job.ID, - "attempt": job.Attempts, - "max_retry": job.MaxRetries, - }) - - return &job, nil -} - -// AckJob marks a job as completed -func (q *RedisQueue) AckJob(ctx context.Context, jobID string) error { - now := time.Now().Unix() - jobKey := fmt.Sprintf("%s%s", JobStatusPrefix, jobID) - - // Update status with TTL for record keeping - if err := q.client.Set(ctx, jobKey, string(StatusCompleted), 48*time.Hour).Err(); err != nil { - q.logger.Warn("Failed to ack job", map[string]interface{}{ - "job_id": jobID, - "error": err.Error(), - }) - } - - // Store completion timestamp - completedKey := fmt.Sprintf("%s:completed:%s", DocumentScraperQueue, jobID) - if err := q.client.Set(ctx, completedKey, now, 48*time.Hour).Err(); err != nil { - q.logger.Warn("Failed to store completion timestamp", map[string]interface{}{ - "job_id": jobID, - "error": err.Error(), - }) - } - - q.logger.Info("Job acknowledged", map[string]interface{}{ - "job_id": jobID, - }) - - return nil -} - -// NackJob marks a job as failed or for retry -func (q *RedisQueue) NackJob(ctx context.Context, jobID string, jobErr error) error { - jobKey := fmt.Sprintf("%s%s", JobStatusPrefix, jobID) - - // Get the job from Redis to check retry attempts - _, err := q.client.Get(ctx, jobKey).Result() - if err != nil && err != redis.Nil { - q.logger.Warn("Failed to get job for nack", map[string]interface{}{ - "job_id": jobID, - "error": err.Error(), - }) - } - - // Check if we should retry by trying to get the job back from processing - // For now, mark as failed - status := StatusFailed - duration := 48 * time.Hour - - if err := q.client.Set(ctx, jobKey, string(status), duration).Err(); err != nil { - q.logger.Warn("Failed to update job status on nack", map[string]interface{}{ - "job_id": jobID, - "error": err.Error(), - }) - } - - // Store error message - errorKey := fmt.Sprintf("%s:error:%s", DocumentScraperQueue, jobID) - if err := q.client.Set(ctx, errorKey, jobErr.Error(), duration).Err(); err != nil { - q.logger.Warn("Failed to store error message", map[string]interface{}{ - "job_id": jobID, - "error": err.Error(), - }) - } - - // Optionally move to DLQ - if q.config.IsDLQEnabled { - jobData, _ := q.client.Get(ctx, JobStatusPrefix+jobID).Result() - if jobData != "" { - if err := q.client.RPush(ctx, DocumentScraperDLQ, jobData).Err(); err != nil { - q.logger.Warn("Failed to add job to DLQ", map[string]interface{}{ - "job_id": jobID, - "error": err.Error(), - }) - } - } - } - - q.logger.Info("Job nacked", map[string]interface{}{ - "job_id": jobID, - "error": jobErr.Error(), - }) - - return nil -} - -// GetJobStatus retrieves the status of a job -func (q *RedisQueue) GetJobStatus(ctx context.Context, jobID string) (JobStatus, error) { - jobKey := fmt.Sprintf("%s%s", JobStatusPrefix, jobID) - status, err := q.client.Get(ctx, jobKey).Result() - if err != nil { - if err == redis.Nil { - return StatusPending, nil - } - return StatusPending, err - } - return JobStatus(status), nil -} - -// GetMetrics returns queue metrics -func (q *RedisQueue) GetMetrics(ctx context.Context) (*QueueMetrics, error) { - metrics := &QueueMetrics{ - UpdatedAt: time.Now().Unix(), - } - - // Get queue size - size, err := q.GetQueueSize(ctx) - if err != nil { - q.logger.Warn("Failed to get queue size for metrics", map[string]interface{}{ - "error": err.Error(), - }) - } - metrics.PendingJobs = size - - // Get DLQ size - dlqSize, err := q.GetDLQSize(ctx) - if err != nil { - q.logger.Warn("Failed to get DLQ size for metrics", map[string]interface{}{ - "error": err.Error(), - }) - } - metrics.FailedJobs = dlqSize - - return metrics, nil -} - -// GetQueueSize returns the number of items in the queue -func (q *RedisQueue) GetQueueSize(ctx context.Context) (int64, error) { - size, err := q.client.LLen(ctx, DocumentScraperQueue).Result() - if err != nil { - return 0, err - } - return size, nil -} - -// GetDLQSize returns the number of items in the dead letter queue -func (q *RedisQueue) GetDLQSize(ctx context.Context) (int64, error) { - size, err := q.client.LLen(ctx, DocumentScraperDLQ).Result() - if err != nil { - return 0, err - } - return size, nil -} - -// GetInProgressJobs returns jobs currently being processed -func (q *RedisQueue) GetInProgressJobs(ctx context.Context) ([]DocumentScraperJob, error) { - // This is a simplified implementation - // In production, you'd track this separately - return nil, nil -} - -// PurgeQueue removes all jobs from the queue -func (q *RedisQueue) PurgeQueue(ctx context.Context) error { - if err := q.client.Del(ctx, DocumentScraperQueue).Err(); err != nil { - q.logger.Error("Failed to purge queue", map[string]interface{}{ - "error": err.Error(), - }) - return err - } - q.logger.Info("Queue purged successfully", map[string]interface{}{}) - return nil -} - -// PurgeDLQ removes all jobs from the dead letter queue -func (q *RedisQueue) PurgeDLQ(ctx context.Context) error { - if err := q.client.Del(ctx, DocumentScraperDLQ).Err(); err != nil { - q.logger.Error("Failed to purge DLQ", map[string]interface{}{ - "error": err.Error(), - }) - return err - } - q.logger.Info("DLQ purged successfully", map[string]interface{}{}) - return nil -} - -// Close closes the Redis connection -func (q *RedisQueue) Close() error { - return q.client.Close() -}