tm_back/cmd/worker/workers/scraper_consumer.go

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
}