From bf3f21fb51a88621f54a6d840a13dd3974242c74 Mon Sep 17 00:00:00 2001 From: "m.nazemi" Date: Sun, 12 Apr 2026 12:29:35 +0330 Subject: [PATCH] agent SDK fix --- cmd/worker/bootstrap/bootstrap.go | 100 +++++- cmd/worker/bootstrap/config.go | 18 +- cmd/worker/workers/notice.go | 6 +- cmd/worker/workers/queue_producer.go | 165 ++++++++++ cmd/worker/workers/scraper_consumer.go | 422 +++++++++++++++++++++++++ pkg/queue/config.go | 53 ++++ pkg/queue/models.go | 85 +++++ pkg/queue/queue.go | 357 +++++++++++++++++++++ 8 files changed, 1186 insertions(+), 20 deletions(-) create mode 100644 cmd/worker/workers/queue_producer.go create mode 100644 cmd/worker/workers/scraper_consumer.go create mode 100644 pkg/queue/config.go create mode 100644 pkg/queue/models.go create mode 100644 pkg/queue/queue.go diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index ce30b69..888eee8 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -13,6 +13,7 @@ import ( "tm/pkg/mongo" "tm/pkg/notification" "tm/pkg/ollama" + "tm/pkg/queue" "tm/pkg/schedule" "tm/pkg/scraper" ) @@ -81,6 +82,8 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger // Debug: Log worker config appLogger.Info("Worker configuration", map[string]interface{}{ "worker_interval": config.Worker.Interval, + "queue_enabled": config.Queue.Enabled, + "queue_mode": config.Queue.Mode, }) // Initialize repositories noticeRepo := notice.NewRepository(mongoManager, appLogger) @@ -91,11 +94,87 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger noticeWorker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService, scraperService) noticeWorker.Run() - // Initialize Document Scraper Worker - // appLogger.Info("Initializing document scraper worker", map[string]interface{}{}) - // scraperWorker := workers.NewDocumentScraperWorker(mongoManager, appLogger, tenderRepo, scraperService) - // appLogger.Info("Starting document scraper worker", map[string]interface{}{}) - // scraperWorker.Run() + // Initialize Queue-based scraping if enabled + 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 Consumer Worker (24/7 continuous processing) + if config.Queue.Enabled && (config.Queue.Mode == "consumer" || config.Queue.Mode == "both") { + appLogger.Info("Starting queue consumer worker", map[string]interface{}{ + "concurrency": config.Queue.Concurrency, + }) + consumerWorker := workers.NewDocumentScraperConsumerWorker( + mongoManager, + appLogger, + tenderRepo, + queueService, + scraperService, + config.Queue.Concurrency, + ) + go consumerWorker.Run() + } else if !config.Queue.Enabled { + appLogger.Info("Queue consumer worker disabled, using scheduled scraping instead", map[string]interface{}{}) + } + + // 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, + }) + + schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo).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, + }) + } + + // Fallback: Keep the old scheduled scraper for backward compatibility + if !config.Queue.Enabled { + appLogger.Info("Scheduling fallback document scraper worker", map[string]interface{}{}) + schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo).AddJob(schedule.Job{ + Name: "Document Scraper Worker Job (Fallback)", + Func: func() { + worker := workers.NewDocumentScraperWorker(mongoManager, appLogger, tenderRepo, scraperService) + worker.Run() + }, + Expr: "0 10 * * * *", // Run at 10 AM every day + }) + } // Initialize Document Summarization Worker (only if MinIO is available) if minioService != nil { @@ -133,17 +212,6 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger }, Expr: workerInterval, }) - - // Schedule Document Scraper Worker job (runs after notice worker) - schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo).AddJob(schedule.Job{ - Name: "Document Scraper Worker Job", - Func: func() { - worker := workers.NewDocumentScraperWorker(mongoManager, appLogger, tenderRepo, scraperService) - worker.Run() - }, - Expr: "0 10 * * * *", // Run at 10 AM every day (1 hour after notice worker) - }) - } // Init Notification Service diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index df79ead..fbb9d05 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -15,7 +15,23 @@ type Config struct { GLM GLMConfig Scraper config.ScraperConfig MinIO config.MinIOConfig - AlertMail string `env:"ALERT_MAIL" envDefault:"alerts@opplens.com"` + 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 { diff --git a/cmd/worker/workers/notice.go b/cmd/worker/workers/notice.go index b370072..8dbbe91 100644 --- a/cmd/worker/workers/notice.go +++ b/cmd/worker/workers/notice.go @@ -573,13 +573,13 @@ func (w *NoticeWorker) getNextSequentialNumber(ctx context.Context, sourceCode, // Create aggregation pipeline to find the highest sequential number pipeline := mongodriver.Pipeline{ - bson.D{{"$match", bson.M{ + bson.D{{Key: "$match", Value: bson.M{ "tender_id": bson.M{"$regex": pattern}, }}}, - bson.D{{"$project", bson.M{ + bson.D{{Key: "$project", Value: bson.M{ "sequential_num": bson.M{"$substr": bson.A{"$tender_id", 5, 3}}, // Extract last 3 chars (NNN) }}}, - bson.D{{"$group", bson.M{ + bson.D{{Key: "$group", Value: bson.M{ "_id": nil, "max_num": bson.M{"$max": bson.M{"$toInt": "$sequential_num"}}, }}}, diff --git a/cmd/worker/workers/queue_producer.go b/cmd/worker/workers/queue_producer.go new file mode 100644 index 0000000..1927860 --- /dev/null +++ b/cmd/worker/workers/queue_producer.go @@ -0,0 +1,165 @@ +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/scraper_consumer.go b/cmd/worker/workers/scraper_consumer.go new file mode 100644 index 0000000..14c6a8f --- /dev/null +++ b/cmd/worker/workers/scraper_consumer.go @@ -0,0 +1,422 @@ +package workers + +import ( + "context" + "fmt" + "time" + tenderPkg "tm/internal/tender" + "tm/pkg/logger" + "tm/pkg/mongo" + "tm/pkg/queue" + "tm/pkg/scraper" +) + +// DocumentScraperConsumerWorker processes scraping jobs from the queue continuously +type DocumentScraperConsumerWorker struct { + Mongo *mongo.ConnectionManager + Logger logger.Logger + TenderRepo tenderPkg.TenderRepository + Queue queue.Queue + Scraper scraper.SDK + Concurrency int + StopChan chan struct{} +} + +// NewDocumentScraperConsumerWorker creates a new consumer worker +func NewDocumentScraperConsumerWorker( + mongo *mongo.ConnectionManager, + logger logger.Logger, + tenderRepo tenderPkg.TenderRepository, + queue queue.Queue, + scraperService scraper.SDK, + concurrency int, +) *DocumentScraperConsumerWorker { + if concurrency <= 0 { + concurrency = 5 // Default concurrency + } + + logger.Info("Creating new document scraper consumer worker", map[string]interface{}{ + "concurrency": concurrency, + }) + + return &DocumentScraperConsumerWorker{ + Mongo: mongo, + Logger: logger, + TenderRepo: tenderRepo, + Queue: queue, + Scraper: scraperService, + Concurrency: concurrency, + StopChan: make(chan struct{}), + } +} + +// Run starts consuming jobs from the queue with specified concurrency +func (w *DocumentScraperConsumerWorker) Run() { + w.Logger.Info("Document scraper consumer worker started", map[string]interface{}{ + "concurrency": w.Concurrency, + }) + + // Create worker pool channels + jobsChan := make(chan *queue.DocumentScraperJob, w.Concurrency) + resultsChan := make(chan *ProcessingResult, w.Concurrency) + + // Start worker goroutines + for i := 0; i < w.Concurrency; i++ { + go w.processJobs(i, jobsChan, resultsChan) + } + + // Start job fetcher + go w.fetchJobs(jobsChan) + + // Start result handler + go w.handleResults(resultsChan) + + // Wait for stop signal + <-w.StopChan + w.Logger.Info("Document scraper consumer worker stopping", map[string]interface{}{}) + + close(jobsChan) + close(resultsChan) +} + +// Stop signals the worker to stop +func (w *DocumentScraperConsumerWorker) Stop() { + w.Logger.Info("Stop signal sent to consumer worker", map[string]interface{}{}) + close(w.StopChan) +} + +// fetchJobs continuously fetches jobs from the queue and sends them to workers +func (w *DocumentScraperConsumerWorker) fetchJobs(jobsChan chan<- *queue.DocumentScraperJob) { + defer func() { + w.Logger.Info("Job fetcher stopped", map[string]interface{}{}) + }() + + for { + select { + case <-w.StopChan: + return + default: + // Try to dequeue a job + ctx, cancel := context.WithTimeout(context.Background(), 35*time.Second) + job, err := w.Queue.Dequeue(ctx) + cancel() + + if err != nil { + w.Logger.Error("Failed to dequeue job", map[string]interface{}{ + "error": err.Error(), + }) + time.Sleep(5 * time.Second) // Back off on error + continue + } + + if job == nil { + // Queue is empty, brief wait before retry + time.Sleep(2 * time.Second) + continue + } + + // Send to processing channel + select { + case jobsChan <- job: + // Job sent successfully + case <-w.StopChan: + return + } + } + } +} + +// processJobs processes jobs from the jobs channel +func (w *DocumentScraperConsumerWorker) processJobs( + workerID int, + jobsChan <-chan *queue.DocumentScraperJob, + resultsChan chan<- *ProcessingResult, +) { + w.Logger.Info("Worker started", map[string]interface{}{ + "worker_id": workerID, + }) + + defer func() { + w.Logger.Info("Worker stopped", map[string]interface{}{ + "worker_id": workerID, + }) + }() + + for { + select { + case job, ok := <-jobsChan: + if !ok { + return // Channel closed + } + + if job == nil { + continue + } + + // Process the job + result := w.processJob(job, workerID) + select { + case resultsChan <- result: + // Result sent successfully + case <-w.StopChan: + return + } + + case <-w.StopChan: + return + } + } +} + +// processJob executes the actual scraping task +func (w *DocumentScraperConsumerWorker) processJob( + job *queue.DocumentScraperJob, + workerID int, +) *ProcessingResult { + startTime := time.Now() + + w.Logger.Info("Processing job", map[string]interface{}{ + "worker_id": workerID, + "job_id": job.ID, + "notice_id": job.NoticeID, + "attempt": job.Attempts, + "max_retries": job.MaxRetries, + }) + + result := &ProcessingResult{ + JobID: job.ID, + WorkerID: workerID, + StartedAt: startTime, + Job: job, + } + + // Skip if missing required fields + if job.NoticeID == "" || job.NoticeURL == "" { + result.Success = false + result.Error = "missing required fields: notice_id or notice_url" + result.CompletedAt = time.Now() + return result + } + + // Call scraper SDK with timeout + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + + scrapeReq := &scraper.ScrapeRequest{ + NoticeID: job.NoticeID, + NoticeURL: job.NoticeURL, + } + + w.Logger.Info("Calling scraper service", map[string]interface{}{ + "job_id": job.ID, + "notice_id": job.NoticeID, + "notice_url": job.NoticeURL, + "worker_id": workerID, + }) + + response, err := w.Scraper.ScrapeDocuments(ctx, scrapeReq) + if err != nil { + result.Success = false + result.Error = err.Error() + result.CompletedAt = time.Now() + result.ShouldRetry = job.Attempts < job.MaxRetries + + w.Logger.Error("Scraping failed", map[string]interface{}{ + "job_id": job.ID, + "worker_id": workerID, + "attempt": job.Attempts, + "error": err.Error(), + "should_retry": result.ShouldRetry, + }) + + return result + } + + if !response.Success { + result.Success = false + result.Error = "scraper returned failure" + if len(response.Errors) > 0 { + result.Error += ": " + response.Errors[0].Error + } + result.CompletedAt = time.Now() + result.ShouldRetry = job.Attempts < job.MaxRetries + + w.Logger.Warn("Scraping reported failure", map[string]interface{}{ + "job_id": job.ID, + "worker_id": workerID, + "error_count": len(response.Errors), + "should_retry": result.ShouldRetry, + }) + + return result + } + + // Update tender with scraped documents + scrapedAt := time.Now().Unix() + tender, err := w.TenderRepo.GetByNoticePublicationID(context.Background(), job.NoticeID) + if err != nil { + result.Success = false + result.Error = "failed to fetch tender: " + err.Error() + result.CompletedAt = time.Now() + result.ShouldRetry = job.Attempts < job.MaxRetries + + w.Logger.Error("Failed to fetch tender for update", map[string]interface{}{ + "job_id": job.ID, + "notice_id": job.NoticeID, + "error": err.Error(), + "worker_id": workerID, + }) + + return result + } + + if tender == nil { + result.Success = false + result.Error = "tender not found" + result.CompletedAt = time.Now() + + w.Logger.Warn("Tender not found for scraping result", map[string]interface{}{ + "job_id": job.ID, + "notice_id": job.NoticeID, + "worker_id": workerID, + }) + + return result + } + + // Update tender with scraped documents + tender.ScrapedDocuments = make([]tenderPkg.ScrapedDocument, len(response.UploadedFiles)) + for i, uploadedFile := range response.UploadedFiles { + tender.ScrapedDocuments[i] = tenderPkg.ScrapedDocument{ + Filename: uploadedFile.Filename, + ObjectName: uploadedFile.ObjectName, + BucketName: "opplens", // Default bucket + Size: uploadedFile.Size, + ScrapedAt: scrapedAt, + } + } + + // Update metadata + tender.ProcessingMetadata.DocumentsScraped = true + tender.ProcessingMetadata.DocumentsScrapedAt = scrapedAt + + // Save to database + if err := w.TenderRepo.Update(context.Background(), tender); err != nil { + result.Success = false + result.Error = "failed to update tender: " + err.Error() + result.CompletedAt = time.Now() + result.ShouldRetry = job.Attempts < job.MaxRetries + + w.Logger.Error("Failed to update tender with scraped documents", map[string]interface{}{ + "job_id": job.ID, + "tender_id": tender.ID.Hex(), + "error": err.Error(), + "worker_id": workerID, + "should_retry": result.ShouldRetry, + }) + + return result + } + + result.Success = true + result.UploadedCount = response.UploadedCount + result.CompletedAt = time.Now() + + w.Logger.Info("Scraping job completed successfully", map[string]interface{}{ + "job_id": job.ID, + "tender_id": tender.ID.Hex(), + "uploaded_count": response.UploadedCount, + "worker_id": workerID, + "duration_seconds": time.Since(startTime).Seconds(), + }) + + return result +} + +// handleResults processes the results from worker jobs +func (w *DocumentScraperConsumerWorker) handleResults(resultsChan <-chan *ProcessingResult) { + defer func() { + w.Logger.Info("Result handler stopped", map[string]interface{}{}) + }() + + for { + select { + case result, ok := <-resultsChan: + if !ok { + return // Channel closed + } + + if result == nil { + continue + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + + if result.Success { + // Acknowledge successful job + if err := w.Queue.AckJob(ctx, result.JobID); err != nil { + w.Logger.Error("Failed to acknowledge job", map[string]interface{}{ + "job_id": result.JobID, + "error": err.Error(), + }) + } + } else { + // Handle failure - potentially retry + if result.ShouldRetry { + w.Logger.Info("Job will be retried", map[string]interface{}{ + "job_id": result.JobID, + "attempt": result.Job.Attempts, + "error": result.Error, + "retry_in": "30 seconds", + }) + + // Re-enqueue the job for retry + job := result.Job + job.NextRetryTime = time.Now().Add(30 * time.Second).Unix() + job.Status = string(queue.StatusRetrying) + + if err := w.Queue.Enqueue(ctx, job); err != nil { + w.Logger.Error("Failed to re-enqueue job for retry", map[string]interface{}{ + "job_id": job.ID, + "error": err.Error(), + }) + // Still nack it if re-enqueue fails + w.Queue.NackJob(ctx, result.JobID, fmt.Errorf(result.Error)) + } + } else { + // Max retries exceeded, send to DLQ + w.Logger.Error("Job exceeded max retries, sending to DLQ", map[string]interface{}{ + "job_id": result.JobID, + "max_retry": result.Job.MaxRetries, + "error": result.Error, + }) + + if err := w.Queue.NackJob(ctx, result.JobID, fmt.Errorf(result.Error)); err != nil { + w.Logger.Error("Failed to nack job", map[string]interface{}{ + "job_id": result.JobID, + "error": err.Error(), + }) + } + } + } + + cancel() + + case <-w.StopChan: + return + } + } +} + +// ProcessingResult represents the result of processing a job +type ProcessingResult struct { + JobID string + WorkerID int + Success bool + Error string + ShouldRetry bool + UploadedCount int + StartedAt time.Time + CompletedAt time.Time + Job *queue.DocumentScraperJob +} diff --git a/pkg/queue/config.go b/pkg/queue/config.go new file mode 100644 index 0000000..43c26b8 --- /dev/null +++ b/pkg/queue/config.go @@ -0,0 +1,53 @@ +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 new file mode 100644 index 0000000..7568533 --- /dev/null +++ b/pkg/queue/models.go @@ -0,0 +1,85 @@ +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 new file mode 100644 index 0000000..4a9ba53 --- /dev/null +++ b/pkg/queue/queue.go @@ -0,0 +1,357 @@ +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() +}