From 086fe0ffd535d955af10caa8d3cdeba7de0a9587 Mon Sep 17 00:00:00 2001 From: Mazyar Date: Mon, 4 May 2026 05:28:56 +0330 Subject: [PATCH] removed filter to get all tenders - bug fix --- cmd/worker/bootstrap/bootstrap.go | 17 +++++++++++++---- cmd/worker/main.go | 7 ++++++- cmd/worker/workers/summarizer.go | 8 +++----- go.mod | 3 --- go.sum | 6 ------ internal/notice/repository.go | 2 +- ted/scraper.go | 8 ++++++++ 7 files changed, 31 insertions(+), 20 deletions(-) diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index 0afab52..f098b54 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -77,7 +77,7 @@ func InitMongoDB(conf config.MongoConfig, log logger.Logger) *mongo.ConnectionMa } // Init Worker -func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, glmService *glm.SDK, minioService *minio.Service) { +func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, glmService *glm.SDK, minioService *minio.Service) *schedule.CronScheduler { // Debug: Log worker config appLogger.Info("Worker configuration", map[string]interface{}{ "worker_interval": config.Worker.Interval, @@ -88,6 +88,9 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger noticeRepo := notice.NewRepository(mongoManager, appLogger) tenderRepo := tender.NewRepository(mongoManager, appLogger) + // Create a single shared cron scheduler for all recurring jobs + scheduler := schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo) + // Initialize Notice Worker appLogger.Info("Starting notice worker", map[string]interface{}{ "processing_limit": config.Worker.NoticeProcessingLimit, @@ -130,7 +133,7 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger "interval": config.Queue.ProducerInterval, }) - schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo).AddJob(schedule.Job{ + scheduler.AddJob(schedule.Job{ Name: "Queue Producer Worker Job", Func: func() { producer := workers.NewQueueProducerWorker( @@ -153,7 +156,7 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger summarizerWorker.Run() // Schedule Document Summarization Worker job - schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo).AddJob(schedule.Job{ + scheduler.AddJob(schedule.Job{ Name: "Document Summarization Worker Job", Func: func() { worker := workers.NewDocumentSummarizationWorker(mongoManager, appLogger, tenderRepo, minioService, glmService) @@ -174,7 +177,7 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger }) } - schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo).AddJob(schedule.Job{ + scheduler.AddJob(schedule.Job{ Name: "Notice Worker Job", Func: func() { worker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService, config.Worker.NoticeProcessingLimit) @@ -182,6 +185,12 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger }, Expr: workerInterval, }) + + // Start the cron scheduler so all recurring jobs actually run + scheduler.Start() + appLogger.Info("Cron scheduler started with all worker jobs", map[string]interface{}{}) + + return scheduler } // Init Notification Service diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 6eb875a..84825d4 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -55,7 +55,7 @@ func main() { } // Initialize Worker - bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, glmService, minioService) + scheduler := bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, glmService, minioService) // Set up signal handling for graceful shutdown signalChan := make(chan os.Signal, 1) @@ -65,5 +65,10 @@ func main() { <-signalChan appLogger.Info("Received shutdown signal, stopping scheduler...", map[string]interface{}{}) + // Stop scheduler gracefully + if scheduler != nil { + scheduler.Stop() + } + appLogger.Info("Worker application stopped", map[string]interface{}{}) } diff --git a/cmd/worker/workers/summarizer.go b/cmd/worker/workers/summarizer.go index b38139f..b75b385 100644 --- a/cmd/worker/workers/summarizer.go +++ b/cmd/worker/workers/summarizer.go @@ -39,9 +39,10 @@ func (w *DocumentSummarizationWorker) Run() { w.Logger.Info("Document summarization worker started", map[string]interface{}{}) limit := 5 // Process fewer tenders at a time since summarization is resource-intensive - skip := 0 for { - tenders, totalCount, err := w.TenderRepo.GetUnSummarizedTenders(context.Background(), limit, skip) + // Always use skip=0 because processed items are excluded by the filter, + // so incrementing skip would cause unprocessed items to be skipped. + tenders, totalCount, err := w.TenderRepo.GetUnSummarizedTenders(context.Background(), limit, 0) if err != nil { w.Logger.Error("Failed to get un-summarized tenders", map[string]interface{}{ "error": err.Error(), @@ -52,7 +53,6 @@ func (w *DocumentSummarizationWorker) Run() { w.Logger.Info("Found tenders for document summarization", map[string]interface{}{ "count": len(tenders), "total_count": totalCount, - "skip": skip, }) if len(tenders) == 0 { @@ -69,8 +69,6 @@ func (w *DocumentSummarizationWorker) Run() { w.summarizeDocumentsForTender(&t) } - - skip += limit } w.Logger.Info("Document summarization worker completed", map[string]interface{}{}) diff --git a/go.mod b/go.mod index 1484f75..003756d 100644 --- a/go.mod +++ b/go.mod @@ -14,7 +14,6 @@ require ( github.com/stretchr/testify v1.10.0 github.com/swaggo/echo-swagger v1.4.1 github.com/swaggo/swag v1.16.6 - go.mongodb.org/mongo-driver v1.17.9 go.mongodb.org/mongo-driver/v2 v2.5.1 go.uber.org/zap v1.26.0 golang.org/x/crypto v0.40.0 @@ -42,7 +41,6 @@ require ( github.com/mailru/easyjson v0.9.0 // indirect github.com/minio/crc64nvme v1.1.0 // indirect github.com/minio/md5-simd v1.1.2 // indirect - github.com/montanaflynn/stats v0.7.1 // indirect github.com/philhofer/fwd v1.2.0 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/rs/xid v1.6.0 // indirect @@ -57,7 +55,6 @@ require ( require ( github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 - github.com/golang/snappy v1.0.0 // indirect github.com/klauspost/compress v1.18.0 // indirect github.com/labstack/gommon v0.4.2 // indirect github.com/mattn/go-colorable v0.1.14 // indirect diff --git a/go.sum b/go.sum index e907bf8..26f27f3 100644 --- a/go.sum +++ b/go.sum @@ -30,8 +30,6 @@ github.com/go-openapi/swag v0.23.1 h1:lpsStH0n2ittzTnbaSloVZLuB5+fvSY/+hnagBjSNZ github.com/go-openapi/swag v0.23.1/go.mod h1:STZs8TbRvEQQKUA+JZNAm3EWlgaOBGpyFDqQnDHMef0= github.com/golang-jwt/jwt/v5 v5.3.0 h1:pv4AsKCKKZuqlgs5sUmn4x8UlGa0kEVt/puTpKx9vvo= github.com/golang-jwt/jwt/v5 v5.3.0/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= -github.com/golang/snappy v1.0.0 h1:Oy607GVXHs7RtbggtPBnr2RmDArIsAefDwvrdWvRhGs= -github.com/golang/snappy v1.0.0/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= @@ -69,8 +67,6 @@ github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34= github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM= github.com/minio/minio-go/v7 v7.0.97 h1:lqhREPyfgHTB/ciX8k2r8k0D93WaFqxbJX36UZq5occ= github.com/minio/minio-go/v7 v7.0.97/go.mod h1:re5VXuo0pwEtoNLsNuSr0RrLfT/MBtohwdaSmPPSRSk= -github.com/montanaflynn/stats v0.7.1 h1:etflOAAHORrCC44V+aR6Ftzort912ZU+YLiSTuV8eaE= -github.com/montanaflynn/stats v0.7.1/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow= github.com/philhofer/fwd v1.2.0 h1:e6DnBTl7vGY+Gz322/ASL4Gyp1FspeMvx1RNDoToZuM= github.com/philhofer/fwd v1.2.0/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= @@ -110,8 +106,6 @@ github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gi github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 h1:ilQV1hzziu+LLM3zUTJ0trRztfwgjqKnBWNtSRkbmwM= github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78/go.mod h1:aL8wCCfTfSfmXjznFBSZNN13rSJjlIOI1fUNAtF7rmI= github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= -go.mongodb.org/mongo-driver v1.17.9 h1:IexDdCuuNJ3BHrELgBlyaH9p60JXAvdzWR128q+U5tU= -go.mongodb.org/mongo-driver v1.17.9/go.mod h1:LlOhpH5NUEfhxcAwG0UEkMqwYcc4JU18gtCdGudk/tQ= go.mongodb.org/mongo-driver/v2 v2.5.1 h1:j2U/Qp+wvueSpqitLCSZPT/+ZpVc1xzuwdHWwl7d8ro= go.mongodb.org/mongo-driver/v2 v2.5.1/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0= go.uber.org/goleak v1.2.0 h1:xqgm/S+aQvhWFTtR0XK3Jvg7z8kGV8P4X14IzwN3Eqk= diff --git a/internal/notice/repository.go b/internal/notice/repository.go index cc489c5..fe04486 100644 --- a/internal/notice/repository.go +++ b/internal/notice/repository.go @@ -147,7 +147,7 @@ func (r *noticeRepository) GetByContractNoticeID(ctx context.Context, contractNo // GetUnProcessedNotices retrieves unprocessed notices func (r *noticeRepository) GetUnProcessedNotices(ctx context.Context, limit int, skip int) ([]Notice, int64, error) { - filter := bson.M{"processing_metadata.processed": false, "main_classification": "72000000"} + filter := bson.M{"processing_metadata.processed": false} pagination := orm.Pagination{ Limit: limit, diff --git a/ted/scraper.go b/ted/scraper.go index 43ea0f5..cc4762d 100644 --- a/ted/scraper.go +++ b/ted/scraper.go @@ -321,6 +321,14 @@ func (s *TEDScraper) processXMLContent(ctx context.Context, content []byte, file // Map to tender entity t := s.mapToTenderFromParsedDoc(notice, "", fileName) + if t == nil { + s.logger.Warn("Skipping document - mapper returned nil (unsupported or incomplete document)", map[string]interface{}{ + "file_name": fileName, + "notice_id": id, + "notice_type": noticeType, + }) + return nil + } t.ContentXML = string(content) // Check if tender already exists to prevent duplicates