tm_back/cmd/worker/workers/queue_producer.go

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()
}