mazyar/tm_back
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

worker notice default limit

Browse Source
This commit is contained in:
m.nazemi
2026-04-15 04:28:51 +03:30
parent a6a1ad51b1
commit 3d651f8596
3 changed files with 16 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/worker/bootstrap/config.go
+1 -1
View File
@@ -36,7 +36,7 @@ type QueueConfig struct {
type WorkerConfig 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"` NoticeProcessingLimit int `env:"WORKER_NOTICE_PROCESSING_LIMIT" envDefault:"0"`
} }
type GLMConfig struct { type GLMConfig struct {
cmd/worker/workers/notice.go
+14 -11
View File
@@ -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 { 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 processingLimit = 5 // Default limit
} }
return &NoticeWorker{ return &NoticeWorker{
@@ -102,13 +102,12 @@ func (w *NoticeWorker) Run() {
w.cleanupProcessedNotices() w.cleanupProcessedNotices()
limit := 10
skip := 0 skip := 0
processedCount := 0 processedCount := 0
maxToProcess := w.ProcessingLimit maxToProcess := w.ProcessingLimit
for processedCount < maxToProcess { for maxToProcess == 0 || processedCount < maxToProcess {
notices, _, err := w.NoticeRepo.GetUnProcessedNotices(context.Background(), limit, skip) notices, _, err := w.NoticeRepo.GetUnProcessedNotices(context.Background(), skip, maxToProcess)
if err != nil { if err != nil {
w.Logger.Error("Failed to get notices", map[string]interface{}{ w.Logger.Error("Failed to get notices", map[string]interface{}{
"error": err.Error(), "error": err.Error(),
@@ -124,6 +123,14 @@ func (w *NoticeWorker) Run() {
} }
for _, n := range notices { 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{}{ w.Logger.Info("Notice", map[string]interface{}{
"notice": n.ID.Hex(), "notice": n.ID.Hex(),
}) })
@@ -179,14 +186,10 @@ func (w *NoticeWorker) Run() {
} }
} }
skip += limit skip += maxToProcess
// Check if we've reached the processing limit // Check if we need to exit outer loop (break from inner loop means we hit the limit)
if processedCount >= maxToProcess { if maxToProcess > 0 && processedCount >= maxToProcess {
w.Logger.Info("Reached maximum processing limit, stopping", map[string]interface{}{
"processed_count": processedCount,
"max_to_process": maxToProcess,
})
break break
} }
} }
internal/notice/repository.go
+1 -1
View File
@@ -16,7 +16,7 @@ type Repository interface {
BulkImport(ctx context.Context, notices []Notice) error BulkImport(ctx context.Context, notices []Notice) error
GetByID(ctx context.Context, id string) (*Notice, error) GetByID(ctx context.Context, id string) (*Notice, error)
GetByContractNoticeID(ctx context.Context, contractNoticeID 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) GetProcessedNotices(ctx context.Context, limit int, skip int) ([]Notice, int64, error)
Delete(ctx context.Context, id string) error Delete(ctx context.Context, id string) error
} }