From 3d651f8596c205200d2d8aa708d4bbca6d57ad83 Mon Sep 17 00:00:00 2001 From: "m.nazemi" Date: Wed, 15 Apr 2026 04:28:51 +0330 Subject: [PATCH] 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 }