diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index 21012c5..df337c3 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -85,8 +85,6 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger "notice_processing_limit": config.Worker.NoticeProcessingLimit, "notice_batch_pause": config.Worker.NoticeBatchPause.String(), "notice_fetch_error_backoff": config.Worker.NoticeFetchErrorBackoff.String(), - "queue_enabled": config.Queue.Enabled, - "queue_mode": config.Queue.Mode, }) // Initialize repositories noticeRepo := notice.NewRepository(mongoManager, appLogger) @@ -95,12 +93,6 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger // Create a single shared cron scheduler for all recurring jobs scheduler := schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo) - if config.Queue.Enabled { - appLogger.Warn("Redis document-scraper queue is deprecated; document scraping is owned by the AI service pipeline", map[string]interface{}{ - "queue_mode": config.Queue.Mode, - }) - } - // Document summarization via external AI service (requires AI_SUMMARIZER_API_BASE_URL) if aiClient != nil { scheduler.AddJob(schedule.Job{ diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index 874631e..47acc1b 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -15,25 +15,9 @@ type Config struct { Scraper config.ScraperConfig MinIO config.MinIOConfig AISummarizer AISummarizerConfig - Queue QueueConfig `env:"QUEUE_" envPrefix:"QUEUE_"` AlertMail string `env:"ALERT_MAIL" envDefault:"alerts@opplens.com"` } -// QueueConfig defines the queue configuration -type QueueConfig struct { - Enabled bool `env:"QUEUE_ENABLED" envDefault:"true"` - RedisURL string `env:"QUEUE_REDIS_URL" envDefault:"redis://localhost:6379"` - Mode string `env:"QUEUE_MODE" envDefault:"consumer"` // "producer", "consumer", or "both" - MaxRetries int `env:"QUEUE_MAX_RETRIES" envDefault:"3"` - RetryBackoff time.Duration `env:"QUEUE_RETRY_BACKOFF" envDefault:"10s"` - JobTimeout time.Duration `env:"QUEUE_JOB_TIMEOUT" envDefault:"5m"` - PoolSize int `env:"QUEUE_POOL_SIZE" envDefault:"10"` - IsDLQEnabled bool `env:"QUEUE_DLQ_ENABLED" envDefault:"true"` - Concurrency int `env:"QUEUE_CONCURRENCY" envDefault:"5"` - ProducerBatch int `env:"QUEUE_PRODUCER_BATCH" envDefault:"100"` - ProducerInterval string `env:"QUEUE_PRODUCER_INTERVAL" envDefault:"0 */6 * * * *"` // Every 6 hours -} - type WorkerConfig struct { Interval string `env:"WORKER_INTERVAL" envDefault:"* 10 * * * *"` NoticeProcessingLimit int `env:"WORKER_NOTICE_PROCESSING_LIMIT" envDefault:"0"` diff --git a/cmd/worker/workers/queue_producer.go b/cmd/worker/workers/queue_producer.go deleted file mode 100644 index 1927860..0000000 --- a/cmd/worker/workers/queue_producer.go +++ /dev/null @@ -1,165 +0,0 @@ -package workers - -import ( - "context" - "time" - "tm/internal/tender" - "tm/pkg/logger" - "tm/pkg/mongo" - "tm/pkg/queue" -) - -// QueueProducerWorker handles enqueuing unscraped tenders to the job queue -type QueueProducerWorker struct { - Mongo *mongo.ConnectionManager - Logger logger.Logger - TenderRepo tender.TenderRepository - Queue queue.Queue - BatchSize int -} - -// NewQueueProducerWorker creates a new queue producer worker -func NewQueueProducerWorker( - mongo *mongo.ConnectionManager, - logger logger.Logger, - tenderRepo tender.TenderRepository, - queue queue.Queue, - batchSize int, -) *QueueProducerWorker { - if batchSize <= 0 { - batchSize = 100 // Default batch size - } - - logger.Info("Creating new queue producer worker", map[string]interface{}{ - "batch_size": batchSize, - }) - - return &QueueProducerWorker{ - Mongo: mongo, - Logger: logger, - TenderRepo: tenderRepo, - Queue: queue, - BatchSize: batchSize, - } -} - -// Run continuously processes and enqueues unscraped tenders -func (w *QueueProducerWorker) Run() { - w.Logger.Info("Queue producer worker started", map[string]interface{}{}) - - defer func() { - w.Logger.Info("Queue producer worker stopped", map[string]interface{}{}) - }() - - batchIndex := 0 - hasMore := true - - for hasMore { - w.Logger.Info("Queue producer fetching batch of tenders", map[string]interface{}{ - "batch_index": batchIndex, - "batch_size": w.BatchSize, - }) - - // Get batch of unscraped tenders - tenders, totalCount, err := w.TenderRepo.GetUnScrapedTenders( - context.Background(), - w.BatchSize, - batchIndex*w.BatchSize, - ) - - if err != nil { - w.Logger.Error("Failed to fetch unscraped tenders", map[string]interface{}{ - "batch_index": batchIndex, - "error": err.Error(), - }) - // Wait before retrying - time.Sleep(30 * time.Second) - continue - } - - w.Logger.Info("Fetched tender batch", map[string]interface{}{ - "batch_size": len(tenders), - "total_count": totalCount, - "batch_index": batchIndex, - }) - - if len(tenders) == 0 { - // No more tenders in this batch - hasMore = false - w.Logger.Info("No more unscraped tenders found", map[string]interface{}{ - "total_processed": totalCount, - }) - break - } - - // Enqueue each tender - successCount := 0 - failCount := 0 - - for _, t := range tenders { - if err := w.enqueueTender(&t); err != nil { - w.Logger.Error("Failed to enqueue tender", map[string]interface{}{ - "tender_id": t.ID.Hex(), - "notice_publication_id": t.NoticePublicationID, - "error": err.Error(), - }) - failCount++ - } else { - successCount++ - } - } - - w.Logger.Info("Batch enqueuing completed", map[string]interface{}{ - "success_count": successCount, - "fail_count": failCount, - "batch_index": batchIndex, - }) - - // Move to next batch - batchIndex++ - - // Check if there are more tenders - if int64(batchIndex*w.BatchSize) >= totalCount { - hasMore = false - } - - // Avoid tight loop - brief pause between batches - time.Sleep(1 * time.Second) - } -} - -// enqueueTender adds a single tender to the queue -func (w *QueueProducerWorker) enqueueTender(t *tender.Tender) error { - // Skip if missing required fields - if t.NoticePublicationID == "" || t.DocumentURI == "" { - w.Logger.Warn("Skipping tender - missing required fields", map[string]interface{}{ - "tender_id": t.ID.Hex(), - "notice_publication_id": t.NoticePublicationID, - "document_uri": t.DocumentURI, - }) - return nil - } - - // Create job - job := queue.NewDocumentScraperJob( - t.NoticePublicationID, - t.DocumentURI, - t.ID.Hex(), - queue.DefaultMaxRetries, - ) - - // Enqueue job - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - if err := w.Queue.Enqueue(ctx, job); err != nil { - return err - } - - return nil -} - -// RunOnce processes all unscraped tenders in a single run (for scheduled use) -func (w *QueueProducerWorker) RunOnce() { - w.Run() -} diff --git a/internal/tender/repository.go b/internal/tender/repository.go index 3b03226..6e9f57a 100644 --- a/internal/tender/repository.go +++ b/internal/tender/repository.go @@ -31,7 +31,6 @@ type TenderRepository interface { GetTenderCountByClassification(ctx context.Context) (map[string]int64, error) Search(ctx context.Context, form *SearchForm, pagination *response.Pagination) (*SearchPageResult, error) GetExpiredTenders(ctx context.Context, beforeDate int64) ([]Tender, error) - GetUnScrapedTenders(ctx context.Context, limit, skip int) ([]Tender, int64, error) GetPendingDocumentScrapeTenders(ctx context.Context, countryCodes []string, minDeadline int64, limit, skip int) ([]Tender, bool, error) GetUnSummarizedTenders(ctx context.Context, limit, skip int) ([]Tender, int64, error) GetUnTranslatedTenders(ctx context.Context, language string, limit, skip int) ([]Tender, int64, error) @@ -549,36 +548,6 @@ func (r *tenderRepository) GetExpiredTenders(ctx context.Context, beforeDate int return result.Items, nil } -// GetUnScrapedTenders retrieves tenders that haven't been processed for document scraping -func (r *tenderRepository) GetUnScrapedTenders(ctx context.Context, limit, skip int) ([]Tender, int64, error) { - filter := bson.M{ - "$or": []bson.M{ - {"processing_metadata.documents_scraped": bson.M{"$exists": false}}, - {"processing_metadata.documents_scraped": false}, - {"processing_metadata.documents_scraped": nil}, - }, - } - - pagination := orm.Pagination{ - Limit: limit, - Skip: skip, - SortField: "created_at", - SortOrder: 1, // Oldest first - } - - result, err := r.ormRepo.FindAll(ctx, filter, pagination) - if err != nil { - r.logger.Error("Failed to get un-scraped tenders", map[string]interface{}{ - "limit": limit, - "skip": skip, - "error": err.Error(), - }) - return nil, 0, err - } - - return result.Items, result.TotalCount, nil -} - // GetPendingDocumentScrapeTenders returns unscraped tenders for the given countries with an active deadline. func (r *tenderRepository) GetPendingDocumentScrapeTenders(ctx context.Context, countryCodes []string, minDeadline int64, limit, skip int) ([]Tender, bool, error) { if len(countryCodes) == 0 { diff --git a/pkg/ai_summarizer/client.go b/pkg/ai_summarizer/client.go index fc1a90d..70c3551 100644 --- a/pkg/ai_summarizer/client.go +++ b/pkg/ai_summarizer/client.go @@ -204,31 +204,11 @@ func (c *Client) FetchAnalyzeOnDemand(ctx context.Context, reqBody AnalyzeReques return &result, nil } -// TriggerPipelineSync calls POST /pipeline/sync. -func (c *Client) TriggerPipelineSync(ctx context.Context) (*PipelineActionResponse, error) { - return c.triggerPipelineAction(ctx, "/pipeline/sync", "pipeline sync") -} - -// TriggerPipelineScrape calls POST /pipeline/scrape. -func (c *Client) TriggerPipelineScrape(ctx context.Context) (*PipelineActionResponse, error) { - return c.triggerPipelineAction(ctx, "/pipeline/scrape", "pipeline scrape") -} - // TriggerPipelineSummarize calls POST /pipeline/summarize. func (c *Client) TriggerPipelineSummarize(ctx context.Context) (*PipelineActionResponse, error) { return c.triggerPipelineAction(ctx, "/pipeline/summarize", "pipeline summarize") } -// TriggerPipelineAnalyze calls POST /pipeline/analyze. -func (c *Client) TriggerPipelineAnalyze(ctx context.Context) (*PipelineActionResponse, error) { - return c.triggerPipelineAction(ctx, "/pipeline/analyze", "pipeline analyze") -} - -// TriggerPipelineDailyRun calls POST /pipeline/daily-run. -func (c *Client) TriggerPipelineDailyRun(ctx context.Context) (*PipelineActionResponse, error) { - return c.triggerPipelineAction(ctx, "/pipeline/daily-run", "pipeline daily-run") -} - func (c *Client) triggerPipelineAction(ctx context.Context, path, label string) (*PipelineActionResponse, error) { url := c.config.APIBaseURL + path httpResp, bodyBytes, err := c.doRawPost(ctx, url, nil) diff --git a/pkg/queue/config.go b/pkg/queue/config.go deleted file mode 100644 index 43c26b8..0000000 --- a/pkg/queue/config.go +++ /dev/null @@ -1,53 +0,0 @@ -package queue - -import ( - "time" -) - -/* -Queue Configuration and Constants - -This package provides a Redis-based job queue system for managing -document scraping tasks in a distributed manner. -*/ - -const ( - // Queue names - DocumentScraperQueue = "scraper:document:queue" - DocumentScraperDLQ = "scraper:document:dlq" // Dead Letter Queue - - // Job status keys - JobStatusPrefix = "job:" - QueueMetricsKey = "queue:metrics" - - // Default timeouts - DefaultJobTimeout = 5 * time.Minute - DefaultMaxRetries = 3 - DefaultRetryBackoff = 10 * time.Second - MaxRetryBackoff = 5 * time.Minute // Cap retry backoff at 5 minutes - InProgressExpiration = 10 * time.Minute // Job must complete within this time -) - -// QueueConfig represents configuration for the queue service -type QueueConfig struct { - RedisURL string - MaxRetries int - RetryBackoff time.Duration - JobTimeout time.Duration - PoolSize int - IsDLQEnabled bool - MaxQueueLength int // 0 means unlimited -} - -// DefaultConfig returns default queue configuration -func DefaultConfig() QueueConfig { - return QueueConfig{ - RedisURL: "redis://localhost:6379", - MaxRetries: DefaultMaxRetries, - RetryBackoff: DefaultRetryBackoff, - JobTimeout: DefaultJobTimeout, - PoolSize: 10, - IsDLQEnabled: true, - MaxQueueLength: 0, // Unlimited by default - } -} diff --git a/pkg/queue/models.go b/pkg/queue/models.go deleted file mode 100644 index 7568533..0000000 --- a/pkg/queue/models.go +++ /dev/null @@ -1,85 +0,0 @@ -package queue - -import ( - "encoding/json" - "time" -) - -// JobStatus represents the status of a queue job -type JobStatus string - -const ( - StatusPending JobStatus = "pending" - StatusInProgress JobStatus = "in_progress" - StatusCompleted JobStatus = "completed" - StatusFailed JobStatus = "failed" - StatusRetrying JobStatus = "retrying" -) - -// DocumentScraperJob represents a document scraping task in the queue -type DocumentScraperJob struct { - ID string `json:"id"` // Unique job ID (typically tender ID) - NoticeID string `json:"notice_id"` // Notice publication ID - NoticeURL string `json:"notice_url"` // Document URL to scrape - TenderID string `json:"tender_id,omitempty"` // Associated tender ID from MongoDB - CreatedAt int64 `json:"created_at"` - UpdatedAt int64 `json:"updated_at"` - Attempts int `json:"attempts"` - MaxRetries int `json:"max_retries"` - Status string `json:"status"` - Error string `json:"error,omitempty"` - NextRetryTime int64 `json:"next_retry_time,omitempty"` - CompletedAt int64 `json:"completed_at,omitempty"` - ProcessingStartedAt int64 `json:"processing_started_at,omitempty"` -} - -// NewDocumentScraperJob creates a new document scraper job -func NewDocumentScraperJob(noticeID, noticeURL, tenderID string, maxRetries int) *DocumentScraperJob { - now := time.Now().Unix() - return &DocumentScraperJob{ - ID: noticeID, - NoticeID: noticeID, - NoticeURL: noticeURL, - TenderID: tenderID, - CreatedAt: now, - UpdatedAt: now, - Attempts: 0, - MaxRetries: maxRetries, - Status: string(StatusPending), - } -} - -// ToJSON marshals the job to JSON -func (j *DocumentScraperJob) ToJSON() (string, error) { - data, err := json.Marshal(j) - return string(data), err -} - -// FromJSON unmarshals JSON to DocumentScraperJob -func (j *DocumentScraperJob) FromJSON(data string) error { - return json.Unmarshal([]byte(data), j) -} - -// QueueMetrics represents queue performance metrics -type QueueMetrics struct { - TotalJobs int64 `json:"total_jobs"` - PendingJobs int64 `json:"pending_jobs"` - InProgressJobs int64 `json:"in_progress_jobs"` - CompletedJobs int64 `json:"completed_jobs"` - FailedJobs int64 `json:"failed_jobs"` - AverageWaitTime time.Duration `json:"average_wait_time"` - AverageProcessTime time.Duration `json:"average_process_time"` - UpdatedAt int64 `json:"updated_at"` -} - -// JobResult represents the result of a completed job -type JobResult struct { - JobID string `json:"job_id"` - Status string `json:"status"` - Success bool `json:"success"` - Error string `json:"error,omitempty"` - CompletedAt int64 `json:"completed_at"` - Duration int64 `json:"duration_ms"` - UploadedCount int `json:"uploaded_count"` - Details map[string]interface{} `json:"details,omitempty"` -} diff --git a/pkg/queue/queue.go b/pkg/queue/queue.go deleted file mode 100644 index 4a9ba53..0000000 --- a/pkg/queue/queue.go +++ /dev/null @@ -1,357 +0,0 @@ -package queue - -import ( - "context" - "fmt" - "time" - "tm/pkg/logger" - - "github.com/redis/go-redis/v9" -) - -// Queue interface defines the queue operations -type Queue interface { - // Job operations - Enqueue(ctx context.Context, job *DocumentScraperJob) error - Dequeue(ctx context.Context) (*DocumentScraperJob, error) - AckJob(ctx context.Context, jobID string) error - NackJob(ctx context.Context, jobID string, err error) error - GetJobStatus(ctx context.Context, jobID string) (JobStatus, error) - - // Queue inspection - GetMetrics(ctx context.Context) (*QueueMetrics, error) - GetQueueSize(ctx context.Context) (int64, error) - GetDLQSize(ctx context.Context) (int64, error) - GetInProgressJobs(ctx context.Context) ([]DocumentScraperJob, error) - - // Cleanup operations - PurgeQueue(ctx context.Context) error - PurgeDLQ(ctx context.Context) error - Close() error -} - -// RedisQueue implements Queue interface using Redis -type RedisQueue struct { - client *redis.Client - config QueueConfig - logger logger.Logger -} - -// NewRedisQueue creates a new Redis-based queue -func NewRedisQueue(config QueueConfig, logger logger.Logger) (Queue, error) { - opt, err := redis.ParseURL(config.RedisURL) - if err != nil { - logger.Error("Failed to parse Redis URL", map[string]interface{}{ - "url": config.RedisURL, - "error": err.Error(), - }) - return nil, fmt.Errorf("failed to parse redis url: %w", err) - } - - opt.PoolSize = config.PoolSize - client := redis.NewClient(opt) - - // Test connection - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - if err := client.Ping(ctx).Err(); err != nil { - logger.Error("Failed to connect to Redis", map[string]interface{}{ - "error": err.Error(), - }) - return nil, fmt.Errorf("failed to connect to redis: %w", err) - } - - logger.Info("Redis queue initialized successfully", map[string]interface{}{ - "pool_size": config.PoolSize, - }) - - return &RedisQueue{ - client: client, - config: config, - logger: logger, - }, nil -} - -// Enqueue adds a job to the queue -func (q *RedisQueue) Enqueue(ctx context.Context, job *DocumentScraperJob) error { - if q.config.MaxQueueLength > 0 { - size, err := q.GetQueueSize(ctx) - if err != nil { - q.logger.Warn("Failed to check queue size", map[string]interface{}{ - "error": err.Error(), - }) - } else if size >= int64(q.config.MaxQueueLength) { - return fmt.Errorf("queue is full: %d/%d", size, q.config.MaxQueueLength) - } - } - - job.CreatedAt = time.Now().Unix() - job.UpdatedAt = time.Now().Unix() - job.Status = string(StatusPending) - - jsonData, err := job.ToJSON() - if err != nil { - q.logger.Error("Failed to marshal job", map[string]interface{}{ - "job_id": job.ID, - "error": err.Error(), - }) - return err - } - - // Push to queue - if err := q.client.RPush(ctx, DocumentScraperQueue, jsonData).Err(); err != nil { - q.logger.Error("Failed to enqueue job", map[string]interface{}{ - "job_id": job.ID, - "error": err.Error(), - }) - return err - } - - // Store job metadata with TTL - jobKey := fmt.Sprintf("%s%s", JobStatusPrefix, job.ID) - if err := q.client.Set(ctx, jobKey, string(StatusPending), 24*time.Hour).Err(); err != nil { - q.logger.Warn("Failed to set job status", map[string]interface{}{ - "job_id": job.ID, - "error": err.Error(), - }) - } - - q.logger.Info("Job enqueued successfully", map[string]interface{}{ - "job_id": job.ID, - "notice_id": job.NoticeID, - }) - - return nil -} - -// Dequeue retrieves and locks the next job from the queue -func (q *RedisQueue) Dequeue(ctx context.Context) (*DocumentScraperJob, error) { - // Use BLPOP to get next job with timeout - result, err := q.client.BLPop(ctx, 30*time.Second, DocumentScraperQueue).Result() - if err != nil { - if err == redis.Nil { - return nil, nil // Queue is empty - } - q.logger.Error("Failed to dequeue job", map[string]interface{}{ - "error": err.Error(), - }) - return nil, err - } - - if len(result) < 2 { - return nil, fmt.Errorf("invalid dequeue result") - } - - jobData := result[1] - - // Parse job - var job DocumentScraperJob - if err := job.FromJSON(jobData); err != nil { - q.logger.Error("Failed to parse job", map[string]interface{}{ - "error": err.Error(), - "data": jobData, - }) - return nil, err - } - - // Mark as in progress with TTL - job.Status = string(StatusInProgress) - job.Attempts++ - job.ProcessingStartedAt = time.Now().Unix() - job.UpdatedAt = time.Now().Unix() - - jobKey := fmt.Sprintf("%s%s", JobStatusPrefix, job.ID) - if err := q.client.Set(ctx, jobKey, string(StatusInProgress), InProgressExpiration).Err(); err != nil { - q.logger.Warn("Failed to update job status", map[string]interface{}{ - "job_id": job.ID, - "error": err.Error(), - }) - } - - q.logger.Info("Job dequeued successfully", map[string]interface{}{ - "job_id": job.ID, - "attempt": job.Attempts, - "max_retry": job.MaxRetries, - }) - - return &job, nil -} - -// AckJob marks a job as completed -func (q *RedisQueue) AckJob(ctx context.Context, jobID string) error { - now := time.Now().Unix() - jobKey := fmt.Sprintf("%s%s", JobStatusPrefix, jobID) - - // Update status with TTL for record keeping - if err := q.client.Set(ctx, jobKey, string(StatusCompleted), 48*time.Hour).Err(); err != nil { - q.logger.Warn("Failed to ack job", map[string]interface{}{ - "job_id": jobID, - "error": err.Error(), - }) - } - - // Store completion timestamp - completedKey := fmt.Sprintf("%s:completed:%s", DocumentScraperQueue, jobID) - if err := q.client.Set(ctx, completedKey, now, 48*time.Hour).Err(); err != nil { - q.logger.Warn("Failed to store completion timestamp", map[string]interface{}{ - "job_id": jobID, - "error": err.Error(), - }) - } - - q.logger.Info("Job acknowledged", map[string]interface{}{ - "job_id": jobID, - }) - - return nil -} - -// NackJob marks a job as failed or for retry -func (q *RedisQueue) NackJob(ctx context.Context, jobID string, jobErr error) error { - jobKey := fmt.Sprintf("%s%s", JobStatusPrefix, jobID) - - // Get the job from Redis to check retry attempts - _, err := q.client.Get(ctx, jobKey).Result() - if err != nil && err != redis.Nil { - q.logger.Warn("Failed to get job for nack", map[string]interface{}{ - "job_id": jobID, - "error": err.Error(), - }) - } - - // Check if we should retry by trying to get the job back from processing - // For now, mark as failed - status := StatusFailed - duration := 48 * time.Hour - - if err := q.client.Set(ctx, jobKey, string(status), duration).Err(); err != nil { - q.logger.Warn("Failed to update job status on nack", map[string]interface{}{ - "job_id": jobID, - "error": err.Error(), - }) - } - - // Store error message - errorKey := fmt.Sprintf("%s:error:%s", DocumentScraperQueue, jobID) - if err := q.client.Set(ctx, errorKey, jobErr.Error(), duration).Err(); err != nil { - q.logger.Warn("Failed to store error message", map[string]interface{}{ - "job_id": jobID, - "error": err.Error(), - }) - } - - // Optionally move to DLQ - if q.config.IsDLQEnabled { - jobData, _ := q.client.Get(ctx, JobStatusPrefix+jobID).Result() - if jobData != "" { - if err := q.client.RPush(ctx, DocumentScraperDLQ, jobData).Err(); err != nil { - q.logger.Warn("Failed to add job to DLQ", map[string]interface{}{ - "job_id": jobID, - "error": err.Error(), - }) - } - } - } - - q.logger.Info("Job nacked", map[string]interface{}{ - "job_id": jobID, - "error": jobErr.Error(), - }) - - return nil -} - -// GetJobStatus retrieves the status of a job -func (q *RedisQueue) GetJobStatus(ctx context.Context, jobID string) (JobStatus, error) { - jobKey := fmt.Sprintf("%s%s", JobStatusPrefix, jobID) - status, err := q.client.Get(ctx, jobKey).Result() - if err != nil { - if err == redis.Nil { - return StatusPending, nil - } - return StatusPending, err - } - return JobStatus(status), nil -} - -// GetMetrics returns queue metrics -func (q *RedisQueue) GetMetrics(ctx context.Context) (*QueueMetrics, error) { - metrics := &QueueMetrics{ - UpdatedAt: time.Now().Unix(), - } - - // Get queue size - size, err := q.GetQueueSize(ctx) - if err != nil { - q.logger.Warn("Failed to get queue size for metrics", map[string]interface{}{ - "error": err.Error(), - }) - } - metrics.PendingJobs = size - - // Get DLQ size - dlqSize, err := q.GetDLQSize(ctx) - if err != nil { - q.logger.Warn("Failed to get DLQ size for metrics", map[string]interface{}{ - "error": err.Error(), - }) - } - metrics.FailedJobs = dlqSize - - return metrics, nil -} - -// GetQueueSize returns the number of items in the queue -func (q *RedisQueue) GetQueueSize(ctx context.Context) (int64, error) { - size, err := q.client.LLen(ctx, DocumentScraperQueue).Result() - if err != nil { - return 0, err - } - return size, nil -} - -// GetDLQSize returns the number of items in the dead letter queue -func (q *RedisQueue) GetDLQSize(ctx context.Context) (int64, error) { - size, err := q.client.LLen(ctx, DocumentScraperDLQ).Result() - if err != nil { - return 0, err - } - return size, nil -} - -// GetInProgressJobs returns jobs currently being processed -func (q *RedisQueue) GetInProgressJobs(ctx context.Context) ([]DocumentScraperJob, error) { - // This is a simplified implementation - // In production, you'd track this separately - return nil, nil -} - -// PurgeQueue removes all jobs from the queue -func (q *RedisQueue) PurgeQueue(ctx context.Context) error { - if err := q.client.Del(ctx, DocumentScraperQueue).Err(); err != nil { - q.logger.Error("Failed to purge queue", map[string]interface{}{ - "error": err.Error(), - }) - return err - } - q.logger.Info("Queue purged successfully", map[string]interface{}{}) - return nil -} - -// PurgeDLQ removes all jobs from the dead letter queue -func (q *RedisQueue) PurgeDLQ(ctx context.Context) error { - if err := q.client.Del(ctx, DocumentScraperDLQ).Err(); err != nil { - q.logger.Error("Failed to purge DLQ", map[string]interface{}{ - "error": err.Error(), - }) - return err - } - q.logger.Info("DLQ purged successfully", map[string]interface{}{}) - return nil -} - -// Close closes the Redis connection -func (q *RedisQueue) Close() error { - return q.client.Close() -}