From fd89371bdf9b0c12bdee775c66efc96f38950af3 Mon Sep 17 00:00:00 2001 From: "m.nazemi" Date: Sat, 11 Apr 2026 18:15:13 +0330 Subject: [PATCH 1/8] tender duplication fix --- ted/scraper.go | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/ted/scraper.go b/ted/scraper.go index 143d3e2..76b011c 100644 --- a/ted/scraper.go +++ b/ted/scraper.go @@ -323,6 +323,24 @@ func (s *TEDScraper) processXMLContent(ctx context.Context, content []byte, file t := s.mapToTenderFromParsedDoc(notice, "", fileName) t.ContentXML = string(content) + // Check if tender already exists to prevent duplicates + existingTender, err := s.findTenderByContractNoticeID(ctx, t.ContractNoticeID) + if err != nil { + s.logger.Error("Failed to check existing tender", map[string]interface{}{ + "contract_notice_id": t.ContractNoticeID, + "error": err.Error(), + }) + return fmt.Errorf("failed to check existing tender: %w", err) + } + + if existingTender != nil { + s.logger.Info("Tender already exists, skipping duplicate", map[string]interface{}{ + "contract_notice_id": t.ContractNoticeID, + "existing_id": existingTender.ID, + }) + return nil + } + // Create new tender s.logger.Info("Creating new tender", map[string]interface{}{ "contract_notice_id": t.ContractNoticeID, From bf3f21fb51a88621f54a6d840a13dd3974242c74 Mon Sep 17 00:00:00 2001 From: "m.nazemi" Date: Sun, 12 Apr 2026 12:29:35 +0330 Subject: [PATCH 2/8] 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() +} From 637db2354b6190b03b7b857673f9a8cd0b0e41ea Mon Sep 17 00:00:00 2001 From: hadi barzegar <83537317+hdbar@users.noreply.github.com> Date: Sun, 12 Apr 2026 12:55:31 +0330 Subject: [PATCH 3/8] add upload script for remote code deployment --- upload.sh | 69 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 69 insertions(+) create mode 100755 upload.sh diff --git a/upload.sh b/upload.sh new file mode 100755 index 0000000..8a8ece5 --- /dev/null +++ b/upload.sh @@ -0,0 +1,69 @@ +#!/bin/bash +set -euo pipefail + +# Remote server configuration +REMOTE_USER="${REMOTE_USER:-root}" +REMOTE_HOST="${REMOTE_HOST:-}" +REMOTE_PATH="${REMOTE_PATH:-/opt/tm-back}" +SSH_KEY="${SSH_KEY:-}" + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" + +usage() { + echo "Usage: $0 --host [--user ] [--path ] [--key ]" + echo "" + echo "Options:" + echo " --host Remote server hostname or IP (required)" + echo " --user Remote user (default: root)" + echo " --path Remote destination path (default: /opt/tm-back)" + echo " --key Path to SSH private key (optional)" + echo "" + echo "Environment variables REMOTE_USER, REMOTE_HOST, REMOTE_PATH, SSH_KEY can also be used." + exit 1 +} + +while [[ $# -gt 0 ]]; do + case "$1" in + --host) REMOTE_HOST="$2"; shift 2 ;; + --user) REMOTE_USER="$2"; shift 2 ;; + --path) REMOTE_PATH="$2"; shift 2 ;; + --key) SSH_KEY="$2"; shift 2 ;; + -h|--help) usage ;; + *) echo "Unknown option: $1"; usage ;; + esac +done + +if [[ -z "$REMOTE_HOST" ]]; then + echo "Error: --host is required" + usage +fi + +SSH_CMD="ssh" +if [[ -n "$SSH_KEY" ]]; then + SSH_CMD="ssh -i $SSH_KEY" +fi + +echo "==> Uploading source code to ${REMOTE_USER}@${REMOTE_HOST}:${REMOTE_PATH}" + +# Create remote directory +${SSH_CMD} "${REMOTE_USER}@${REMOTE_HOST}" "mkdir -p ${REMOTE_PATH}" + +# Rsync source code, excluding non-essential files +rsync -avz --progress \ + --exclude 'bin/' \ + --exclude 'artifacts/' \ + --exclude '.git/' \ + --exclude '.idea/' \ + --exclude '.vscode/' \ + --exclude '*.exe' \ + --exclude 'Dockerfile' \ + --exclude 'upload.sh' \ + ${SSH_KEY:+-e "ssh -i $SSH_KEY"} \ + "${SCRIPT_DIR}/" \ + "${REMOTE_USER}@${REMOTE_HOST}:${REMOTE_PATH}/" + +echo "" +echo "==> Upload complete to ${REMOTE_USER}@${REMOTE_HOST}:${REMOTE_PATH}" +echo "" +echo "To build and run on the remote server:" +echo " ${SSH_CMD} ${REMOTE_USER}@${REMOTE_HOST} 'cd ${REMOTE_PATH} && go build -o bin/web ./cmd/web'" From 07320c58b08e709c67ebbc4c6565835521687ce4 Mon Sep 17 00:00:00 2001 From: "m.nazemi" Date: Sun, 12 Apr 2026 15:11:37 +0330 Subject: [PATCH 4/8] scraper SSL fix --- cmd/scraper/Dockerfile | 2 +- ted/calendar.go | 4 ++-- ted/scraper.go | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/cmd/scraper/Dockerfile b/cmd/scraper/Dockerfile index 4cb0468..3fe0d61 100644 --- a/cmd/scraper/Dockerfile +++ b/cmd/scraper/Dockerfile @@ -9,7 +9,7 @@ FROM docker-mirror.ravanertebat.ir/base/amd64/bash:$BASH_VERSION # RUN apk update # # RUN apk add --upgrade busybox-suid # RUN apk --no-cache add --update busybox-suid \ -# && apk --no-cache add curl rsync tzdata tcpdump tree ca-certificates s-nail \ +# && apk --no-cache add --upgrade ca-certificates curl rsync tzdata tcpdump tree s-nail \ # && ln -s /usr/bin/mail /usr/bin/s-nail \ # && mkdir -p /go/bin/media \ # && mkdir -p /go/bin/log \ diff --git a/ted/calendar.go b/ted/calendar.go index 4491b1f..0015204 100644 --- a/ted/calendar.go +++ b/ted/calendar.go @@ -14,13 +14,13 @@ const ( ) // Get OJS Number -func GetOJS(baseURL string, year int, date string) (string, error) { +func GetOJS(baseURL string, year int, date string, client *http.Client) (string, error) { var ( ojs string ) url := fmt.Sprintf(calendarUrl, baseURL, year) - resp, err := http.Get(url) + resp, err := client.Get(url) if err != nil { return ojs, err } diff --git a/ted/scraper.go b/ted/scraper.go index 76b011c..43ea0f5 100644 --- a/ted/scraper.go +++ b/ted/scraper.go @@ -463,7 +463,7 @@ func (s *TEDScraper) processSingleDate(ctx context.Context, date time.Time) erro }) // get ojs - ojs, err := GetOJS(s.config.BaseURL, date.Year(), date.Format("02/01/2006")) + ojs, err := GetOJS(s.config.BaseURL, date.Year(), date.Format("02/01/2006"), s.httpClient) if err != nil { s.logger.Error("Failed to get OJS", map[string]interface{}{ "date": date.Format("02/01/2006"), From a6a1ad51b1fac95c971b695694026b103b4c24cf Mon Sep 17 00:00:00 2001 From: "m.nazemi" Date: Sun, 12 Apr 2026 21:03:34 +0330 Subject: [PATCH 5/8] worker notice limit config --- cmd/worker/bootstrap/bootstrap.go | 8 ++++--- cmd/worker/bootstrap/config.go | 3 ++- cmd/worker/workers/notice.go | 37 ++++++++++++++++++------------- 3 files changed, 28 insertions(+), 20 deletions(-) diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index 888eee8..9768096 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -90,8 +90,10 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger tenderRepo := tender.NewRepository(mongoManager, appLogger) // Initialize Notice Worker - appLogger.Info("Starting notice worker", map[string]interface{}{}) - noticeWorker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService, scraperService) + appLogger.Info("Starting notice worker", map[string]interface{}{ + "processing_limit": config.Worker.NoticeProcessingLimit, + }) + noticeWorker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService, scraperService, config.Worker.NoticeProcessingLimit) noticeWorker.Run() // Initialize Queue-based scraping if enabled @@ -207,7 +209,7 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo).AddJob(schedule.Job{ Name: "Notice Worker Job", Func: func() { - worker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService, scraperService) + worker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService, scraperService, config.Worker.NoticeProcessingLimit) worker.Run() }, Expr: workerInterval, diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index fbb9d05..08358c8 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -35,7 +35,8 @@ type QueueConfig struct { } type WorkerConfig struct { - Interval string `env:"WORKER_INTERVAL" envDefault:"* 10 * * * *"` + Interval string `env:"WORKER_INTERVAL" envDefault:"* 10 * * * *"` + NoticeProcessingLimit int `env:"WORKER_NOTICE_PROCESSING_LIMIT" envDefault:"5"` } type GLMConfig struct { diff --git a/cmd/worker/workers/notice.go b/cmd/worker/workers/notice.go index 8dbbe91..119b4e3 100644 --- a/cmd/worker/workers/notice.go +++ b/cmd/worker/workers/notice.go @@ -18,24 +18,29 @@ import ( ) type NoticeWorker struct { - Mongo *mongo.ConnectionManager - Logger logger.Logger - Notify *notification.SDK - NoticeRepo notice.Repository - TenderRepo tender.TenderRepository - GLM *glm.SDK - Scraper scraper.SDK + Mongo *mongo.ConnectionManager + Logger logger.Logger + Notify *notification.SDK + NoticeRepo notice.Repository + TenderRepo tender.TenderRepository + GLM *glm.SDK + Scraper scraper.SDK + ProcessingLimit int } -func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify *notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository, glmService *glm.SDK, scraper scraper.SDK) *NoticeWorker { +func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify *notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository, glmService *glm.SDK, scraper scraper.SDK, processingLimit int) *NoticeWorker { + if processingLimit <= 0 { + processingLimit = 5 // Default limit + } return &NoticeWorker{ - Mongo: mongo, - Logger: logger, - Notify: notify, - NoticeRepo: noticeRepo, - TenderRepo: tenderRepo, - GLM: glmService, - Scraper: scraper, + Mongo: mongo, + Logger: logger, + Notify: notify, + NoticeRepo: noticeRepo, + TenderRepo: tenderRepo, + GLM: glmService, + Scraper: scraper, + ProcessingLimit: processingLimit, } } @@ -100,7 +105,7 @@ func (w *NoticeWorker) Run() { limit := 10 skip := 0 processedCount := 0 - maxToProcess := 5 // Limit processing to allow scraper worker to run + maxToProcess := w.ProcessingLimit for processedCount < maxToProcess { notices, _, err := w.NoticeRepo.GetUnProcessedNotices(context.Background(), limit, skip) From 3d651f8596c205200d2d8aa708d4bbca6d57ad83 Mon Sep 17 00:00:00 2001 From: "m.nazemi" Date: Wed, 15 Apr 2026 04:28:51 +0330 Subject: [PATCH 6/8] worker notice default limit --- cmd/worker/bootstrap/config.go | 2 +- cmd/worker/workers/notice.go | 25 ++++++++++++++----------- internal/notice/repository.go | 2 +- 3 files changed, 16 insertions(+), 13 deletions(-) diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index 08358c8..3e8c27a 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -36,7 +36,7 @@ type QueueConfig struct { type WorkerConfig struct { Interval string `env:"WORKER_INTERVAL" envDefault:"* 10 * * * *"` - NoticeProcessingLimit int `env:"WORKER_NOTICE_PROCESSING_LIMIT" envDefault:"5"` + NoticeProcessingLimit int `env:"WORKER_NOTICE_PROCESSING_LIMIT" envDefault:"0"` } type GLMConfig struct { diff --git a/cmd/worker/workers/notice.go b/cmd/worker/workers/notice.go index 119b4e3..b7c9b95 100644 --- a/cmd/worker/workers/notice.go +++ b/cmd/worker/workers/notice.go @@ -29,7 +29,7 @@ type NoticeWorker struct { } func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify *notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository, glmService *glm.SDK, scraper scraper.SDK, processingLimit int) *NoticeWorker { - if processingLimit <= 0 { + if processingLimit < 0 { processingLimit = 5 // Default limit } return &NoticeWorker{ @@ -102,13 +102,12 @@ func (w *NoticeWorker) Run() { w.cleanupProcessedNotices() - limit := 10 skip := 0 processedCount := 0 maxToProcess := w.ProcessingLimit - for processedCount < maxToProcess { - notices, _, err := w.NoticeRepo.GetUnProcessedNotices(context.Background(), limit, skip) + for maxToProcess == 0 || processedCount < maxToProcess { + notices, _, err := w.NoticeRepo.GetUnProcessedNotices(context.Background(), skip, maxToProcess) if err != nil { w.Logger.Error("Failed to get notices", map[string]interface{}{ "error": err.Error(), @@ -124,6 +123,14 @@ func (w *NoticeWorker) Run() { } for _, n := range notices { + if maxToProcess > 0 && processedCount >= maxToProcess { + w.Logger.Info("Reached maximum processing limit, stopping", map[string]interface{}{ + "processed_count": processedCount, + "max_to_process": maxToProcess, + }) + break + } + w.Logger.Info("Notice", map[string]interface{}{ "notice": n.ID.Hex(), }) @@ -179,14 +186,10 @@ func (w *NoticeWorker) Run() { } } - skip += limit + skip += maxToProcess - // Check if we've reached the processing limit - if processedCount >= maxToProcess { - w.Logger.Info("Reached maximum processing limit, stopping", map[string]interface{}{ - "processed_count": processedCount, - "max_to_process": maxToProcess, - }) + // Check if we need to exit outer loop (break from inner loop means we hit the limit) + if maxToProcess > 0 && processedCount >= maxToProcess { break } } diff --git a/internal/notice/repository.go b/internal/notice/repository.go index f6ccfa4..cc489c5 100644 --- a/internal/notice/repository.go +++ b/internal/notice/repository.go @@ -16,7 +16,7 @@ type Repository interface { BulkImport(ctx context.Context, notices []Notice) error GetByID(ctx context.Context, id string) (*Notice, error) GetByContractNoticeID(ctx context.Context, contractNoticeID string) (*Notice, error) - GetUnProcessedNotices(ctx context.Context, limit int, skip int) ([]Notice, int64, error) + GetUnProcessedNotices(ctx context.Context, ProcessingLimit int, skip int) ([]Notice, int64, error) GetProcessedNotices(ctx context.Context, limit int, skip int) ([]Notice, int64, error) Delete(ctx context.Context, id string) error } From 7145977250b533218ce82ab4be0149b4ff18d189 Mon Sep 17 00:00:00 2001 From: "m.nazemi" Date: Wed, 15 Apr 2026 15:53:51 +0330 Subject: [PATCH 7/8] get notification fix --- internal/notification/handler.go | 11 +++++++++++ internal/notification/service.go | 18 +++++++++++++++--- 2 files changed, 26 insertions(+), 3 deletions(-) diff --git a/internal/notification/handler.go b/internal/notification/handler.go index 678f2ee..706e2f8 100644 --- a/internal/notification/handler.go +++ b/internal/notification/handler.go @@ -259,6 +259,17 @@ func (h *NotificationHandler) CustomerNotifications(c echo.Context) error { // Call service notifications, err := h.service.GetNotifications(c.Request().Context(), form, pagination) if err != nil { + // Log the actual error for debugging + c.Logger().Error("GetNotifications failed", map[string]interface{}{ + "error": err.Error(), + "status": form.Status, + "priority": form.Priority, + "eventType": form.EventType, + "recipient": form.Recipient, + "limit": pagination.Limit, + "offset": pagination.Offset, + "page": (pagination.Offset / pagination.Limit) + 1, + }) return response.InternalServerError(c, "Failed to get notifications") } diff --git a/internal/notification/service.go b/internal/notification/service.go index ed8821a..3158b46 100644 --- a/internal/notification/service.go +++ b/internal/notification/service.go @@ -201,7 +201,7 @@ func (s *notificationService) GetNotifications(ctx context.Context, req *SearchF Seen: req.Seen, Priority: req.Priority, PerPage: pagination.Limit, - Page: pagination.Offset / pagination.Limit, + Page: (pagination.Offset / pagination.Limit) + 1, // Convert 0-based offset to 1-based page number }) if err != nil { s.logger.Error("Failed to get notifications", map[string]interface{}{ @@ -233,13 +233,25 @@ func (s *notificationService) GetNotifications(ctx context.Context, req *SearchF recipient = cust } + // Handle Link pointer - return nil if empty + var linkPtr *string + if notification.Link != "" { + linkPtr = ¬ification.Link + } + + // Handle Image pointer - return nil if empty + var imagePtr *string + if notification.Image != "" { + imagePtr = ¬ification.Image + } + notificationsResponse = append(notificationsResponse, &NotificationResponse{ ID: notification.ID, UserID: notification.UserID, Recipient: recipient, Title: notification.Title, Message: notification.Message, - Link: ¬ification.Link, + Link: linkPtr, Priority: notification.Priority, Type: notification.Type, EventType: notification.EventType, @@ -248,7 +260,7 @@ func (s *notificationService) GetNotifications(ctx context.Context, req *SearchF Metadata: notification.Metadata, ScheduleAt: notification.ScheduledAt, IsScheduled: notification.IsScheduled, - Image: ¬ification.Image, + Image: imagePtr, Seen: notification.Seen, SeenAt: notification.SeenAt, ScheduledAt: notification.ScheduledAt, From 6c76b7bcac630fe81bb2a2bcd2cf4a9e795795af Mon Sep 17 00:00:00 2001 From: hadi barzegar <83537317+hdbar@users.noreply.github.com> Date: Thu, 16 Apr 2026 11:56:01 +0330 Subject: [PATCH 8/8] enhance tender mapping with additional notice fields and processing metadata --- cmd/worker/workers/notice.go | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/cmd/worker/workers/notice.go b/cmd/worker/workers/notice.go index b7c9b95..260d722 100644 --- a/cmd/worker/workers/notice.go +++ b/cmd/worker/workers/notice.go @@ -474,6 +474,23 @@ func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) { t.MainClassification = n.MainClassification t.AdditionalClassifications = n.AdditionalClassifications t.OfficialLanguages = n.OfficialLanguages + t.Status = tender.TenderStatus(n.Status) + t.Source = tender.TenderSource(n.Source) + t.ContractNoticeID = n.ContractNoticeID + t.ContractFolderID = n.ContractFolderID + t.SourceFileURL = n.SourceFileURL + t.SourceFileName = n.SourceFileName + // Preserve document-scraping/summarization flags on the tender (managed by + // downstream workers), but refresh the notice-side processing metadata. + t.ProcessingMetadata.ScrapedAt = n.ProcessingMetadata.ScrapedAt + t.ProcessingMetadata.ProcessedAt = n.ProcessingMetadata.ProcessedAt + t.ProcessingMetadata.ProcessingVersion = n.ProcessingMetadata.ProcessingVersion + t.ProcessingMetadata.ParsingErrors = n.ProcessingMetadata.ParsingErrors + t.ProcessingMetadata.ValidationErrors = n.ProcessingMetadata.ValidationErrors + t.ProcessingMetadata.EnrichmentData = n.ProcessingMetadata.EnrichmentData + t.ProcessingMetadata.TranslatedData = n.ProcessingMetadata.TranslatedData + t.ProcessingMetadata.TranslatedAt = n.ProcessingMetadata.TranslatedAt + t.ProcessingMetadata.Processed = n.ProcessingMetadata.Processed t.CancellationReason = n.CancellationReason t.CancellationDate = n.CancellationDate t.AwardDate = n.AwardDate