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..df337c3 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, @@ -87,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) @@ -97,70 +93,18 @@ 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, - }) - } - - // 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 +339,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/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/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/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/cmd/worker/workers/summarizer.go b/cmd/worker/workers/summarizer.go index 991d917..dcc352f 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,28 +46,31 @@ 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 + 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(), }) - continue + return } w.Logger.Info("Found tenders for document summarization", map[string]interface{}{ "count": len(tenders), "total_count": totalCount, + "skip": skip, }) if len(tenders) == 0 { - w.Logger.Info("No more tenders to process for document summarization", map[string]interface{}{}) break } @@ -72,78 +79,85 @@ 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) + } + + skip += len(tenders) + if int64(skip) >= totalCount { + break } } 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(), }) - t.ProcessingMetadata.DocumentsSummarized = true - t.ProcessingMetadata.DocumentsSummarizedAt = time.Now().Unix() - _ = w.updateTenderSummarizationStatus(t) + 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 } - 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 +166,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/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/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..dc092c3 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" @@ -589,12 +589,58 @@ 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") } +// 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 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/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/internal/tender/service.go b/internal/tender/service.go index a3e7c02..840c9e2 100644 --- a/internal/tender/service.go +++ b/internal/tender/service.go @@ -15,13 +15,15 @@ import ( "tm/internal/customer" "tm/pkg/ai_summarizer" "tm/pkg/logger" + orm "tm/pkg/mongo" "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 +32,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 +63,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 +455,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 +469,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. @@ -1147,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 @@ -1161,37 +1202,27 @@ 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 } - 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 +1278,95 @@ 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, ErrEmptyTenderID + } + if s.aiSummarizerClient == nil { + return nil, ErrAINotConfigured + } + + 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(), + }) + 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, ErrTenderNotFound + } + if t.NoticePublicationID == "" { + return nil, ErrTenderMissingNoticeID + } + if strings.TrimSpace(t.ContractFolderID) == "" { + return nil, ErrTenderMissingContractFolderID + } + + 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 +1630,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 +1649,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..70c3551 100644 --- a/pkg/ai_summarizer/client.go +++ b/pkg/ai_summarizer/client.go @@ -169,6 +169,98 @@ 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 +} + +// TriggerPipelineSummarize calls POST /pipeline/summarize. +func (c *Client) TriggerPipelineSummarize(ctx context.Context) (*PipelineActionResponse, error) { + return c.triggerPipelineAction(ctx, "/pipeline/summarize", "pipeline summarize") +} + +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. 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() -}