From 1c3ad648c14931bf8fce7ecf7be0e91ac1b1ca06 Mon Sep 17 00:00:00 2001 From: "n.nakhostin" Date: Sun, 5 Oct 2025 16:17:41 +0330 Subject: [PATCH] Implement Worker Service and Refactor Tender Initialization - Introduced a new worker service with a dedicated main entry point for handling background tasks, including MongoDB connection management and notification service initialization. - Added a bootstrap package to manage application configuration, logging, and service initialization for the worker. - Implemented a NoticeWorker to process unprocessed notices and create corresponding tender entities, enhancing the integration of notice management within the tender system. - Refactored the tender service to remove the Ollama SDK dependency, streamlining the service initialization in the main application. - Enhanced the notice repository with a method to retrieve unprocessed notices, improving data handling capabilities. --- cmd/web/main.go | 5 +- cmd/worker/bootstrap/bootstrap.go | 147 +++++++++++++++ cmd/worker/bootstrap/config.go | 18 ++ cmd/worker/main.go | 55 ++++++ cmd/worker/workers/notice.go | 300 ++++++++++++++++++++++++++++++ internal/notice/repository.go | 22 +++ internal/tender/service.go | 5 +- 7 files changed, 544 insertions(+), 8 deletions(-) create mode 100644 cmd/worker/bootstrap/bootstrap.go create mode 100644 cmd/worker/bootstrap/config.go create mode 100644 cmd/worker/main.go create mode 100644 cmd/worker/workers/notice.go diff --git a/cmd/web/main.go b/cmd/web/main.go index f300820..c03f5ab 100644 --- a/cmd/web/main.go +++ b/cmd/web/main.go @@ -129,9 +129,6 @@ func main() { // Initialize notification service notificationSDK := bootstrap.InitNotificationService(conf.Notification, logger) - // Initialize ollama service - ollamaSDK := bootstrap.InitOllamaService(conf.Ollama, logger) - // Initialize authorization service userAuthService := bootstrap.InitAuthorizationService(conf.UserAuth.JWT, redisClient, logger) customerAuthService := bootstrap.InitAuthorizationService(conf.CustomerAuth.JWT, redisClient, logger) @@ -158,7 +155,7 @@ func main() { categoryService := company_category.NewService(categoryRepository, logger) companyService := company.NewService(companyRepository, categoryService, logger) customerService := customer.NewService(customerRepository, logger, customerAuthService, companyService, redisClient, notificationSDK, customerValidator) - tenderService := tender.NewService(tenderRepository, companyService, ollamaSDK, logger) + tenderService := tender.NewService(tenderRepository, companyService, logger) feedbackService := feedback.NewService(feedbackRepo, tenderService, companyService, logger) tenderApprovalService := tender_approval.NewService(tenderApprovalRepository, tenderService, logger) inquiryService := inquiry.NewService(inquiryRepository, notificationSDK, logger) diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go new file mode 100644 index 0000000..96fbea5 --- /dev/null +++ b/cmd/worker/bootstrap/bootstrap.go @@ -0,0 +1,147 @@ +package bootstrap + +import ( + "fmt" + "time" + "tm/cmd/worker/workers" + "tm/internal/notice" + "tm/internal/tender" + "tm/pkg/config" + "tm/pkg/logger" + "tm/pkg/mongo" + "tm/pkg/notification" + "tm/pkg/ollama" + "tm/pkg/schedule" +) + +// Init Application Configuration +func InitConfig() (*Config, error) { + conf, err := config.LoadConfig(".", &Config{}) + if err != nil { + return nil, fmt.Errorf("failed to load config: %v", err) + } + + return conf, nil +} + +// Init Logger Service +func InitLogger(conf config.LoggingConfig) logger.Logger { + return logger.NewLogger(&logger.Config{ + Level: conf.Level, + Format: conf.Format, + Output: conf.Output, + File: logger.FileConfig{ + Path: conf.File.Path, + MaxSize: conf.File.MaxSize, + MaxBackups: conf.File.MaxBackups, + MaxAge: conf.File.MaxAge, + Compress: conf.File.Compress, + }, + }) +} + +// Init MongoDB Connection Manager +func InitMongoDB(conf config.MongoConfig, log logger.Logger) *mongo.ConnectionManager { + // Convert infra.MongoConfig to mongo.ConnectionConfig + connectionConfig := mongo.ConnectionConfig{ + URI: conf.URI, + Database: conf.Name, + MaxPoolSize: uint64(conf.MaxPoolSize), + MinPoolSize: 5, // Default minimum pool size + MaxConnIdleTime: 30 * time.Minute, // Default idle time + ConnectTimeout: conf.Timeout, + ServerSelectionTimeout: 5 * time.Second, // Default server selection timeout + } + + // Create MongoDB connection manager + connectionManager, err := mongo.NewConnectionManager(connectionConfig, log) + if err != nil { + log.Error("Failed to initialize MongoDB connection", map[string]interface{}{ + "error": err.Error(), + "uri": conf.URI, + "db": conf.Name, + }) + panic(fmt.Sprintf("Failed to initialize MongoDB connection: %v", err)) + } + + log.Info("MongoDB connection manager initialized successfully", map[string]interface{}{ + "database": conf.Name, + "uri": conf.URI, + "pool_size": conf.MaxPoolSize, + }) + + return connectionManager +} + +// Init Worker +func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, ollamaSDK *ollama.SDK) { + // Initialize repositories + noticeRepo := notice.NewRepository(mongoManager, appLogger) + tenderRepo := tender.NewRepository(mongoManager, appLogger) + + // start Worker job + schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo).AddJob(schedule.Job{ + Name: "Worker Job", + Func: func() { + worker := workers.NewNoticeWorker(mongoManager, appLogger, notify, noticeRepo, tenderRepo, ollamaSDK) + worker.Run() + }, + Expr: config.Worker.Interval, + }) + +} + +// Init Notification Service +func InitNotificationService(conf config.NotificationConfig, log logger.Logger) notification.SDK { + notificationConfig := ¬ification.Config{ + BaseURL: conf.BaseURL, + Timeout: conf.Timeout, + RetryAttempts: conf.RetryAttempts, + RetryDelay: conf.RetryDelay, + EnableLogging: conf.EnableLogging, + UserAgent: conf.UserAgent, + } + + notificationSDK, err := notification.New(notificationConfig, log) + if err != nil { + log.Error("Failed to initialize Notification SDK", map[string]interface{}{ + "error": err.Error(), + }) + fmt.Printf("Failed to initialize Notification SDK: %v\n", err.Error()) + } + + log.Info("Notification SDK initialized successfully", map[string]interface{}{ + "base_url": conf.BaseURL, + "timeout": conf.Timeout, + "retry_attempts": conf.RetryAttempts, + "retry_delay": conf.RetryDelay, + "enable_logging": conf.EnableLogging, + "user_agent": conf.UserAgent, + }) + + return *notificationSDK +} + +// Init Ollama Service +func InitOllamaService(conf config.OllamaConfig, log logger.Logger) *ollama.SDK { + ollamaSDK, err := ollama.New(&ollama.Config{ + BaseURL: conf.BaseURL, + Timeout: conf.Timeout, + RetryAttempts: conf.RetryAttempts, + RetryDelay: conf.RetryDelay, + EnableLogging: conf.EnableLogging, + UserAgent: conf.UserAgent, + DefaultModel: conf.DefaultModel, + DefaultTemperature: conf.DefaultTemperature, + DefaultTopP: conf.DefaultTopP, + DefaultMaxTokens: conf.DefaultMaxTokens, + EnableStreaming: conf.EnableStreaming, + }, log) + if err != nil { + log.Error("Failed to initialize Ollama SDK", map[string]interface{}{ + "error": err.Error(), + }) + } + + return ollamaSDK +} diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go new file mode 100644 index 0000000..7d219ff --- /dev/null +++ b/cmd/worker/bootstrap/config.go @@ -0,0 +1,18 @@ +package bootstrap + +import ( + "tm/pkg/config" +) + +// Config defines the scraper application configuration +type Config struct { + Database config.DatabaseConfig + Logging config.LoggingConfig + Notification config.NotificationConfig + Ollama config.OllamaConfig + Worker WorkerConfig +} + +type WorkerConfig struct { + Interval string `env:"WORKER_INTERVAL" envDefault:"* 10 * * * *"` +} diff --git a/cmd/worker/main.go b/cmd/worker/main.go new file mode 100644 index 0000000..8fdb9ee --- /dev/null +++ b/cmd/worker/main.go @@ -0,0 +1,55 @@ +package main + +import ( + "context" + "os" + "os/signal" + "syscall" + "time" + "tm/cmd/worker/bootstrap" +) + +func main() { + // Load configuration + config, err := bootstrap.InitConfig() + if err != nil { + panic(err) + } + + // Initialize logger + appLogger := bootstrap.InitLogger(config.Logging) + appLogger.Info("Starting worker application", map[string]interface{}{ + "version": "1.0.0", + }) + + // Initialize MongoDB connection + mongoManager := bootstrap.InitMongoDB(config.Database.MongoDB, appLogger) + defer func() { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := mongoManager.Close(ctx); err != nil { + appLogger.Error("Failed to close MongoDB connection", map[string]interface{}{ + "error": err.Error(), + }) + } + }() + + // Initialize notification service + notificationService := bootstrap.InitNotificationService(config.Notification, appLogger) + + // Initialize ollama service + ollamaSDK := bootstrap.InitOllamaService(config.Ollama, appLogger) + + // Initialize Worker + bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, ollamaSDK) + + // Set up signal handling for graceful shutdown + signalChan := make(chan os.Signal, 1) + signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM) + + // Wait for shutdown signal + <-signalChan + appLogger.Info("Received shutdown signal, stopping scheduler...", map[string]interface{}{}) + + appLogger.Info("Worker application stopped", map[string]interface{}{}) +} diff --git a/cmd/worker/workers/notice.go b/cmd/worker/workers/notice.go new file mode 100644 index 0000000..977480b --- /dev/null +++ b/cmd/worker/workers/notice.go @@ -0,0 +1,300 @@ +package workers + +import ( + "context" + "fmt" + "tm/internal/notice" + "tm/internal/tender" + "tm/pkg/logger" + "tm/pkg/mongo" + "tm/pkg/notification" + "tm/pkg/ollama" +) + +type NoticeWorker struct { + Mongo *mongo.ConnectionManager + Logger logger.Logger + Notify notification.SDK + Ollama *ollama.SDK + NoticeRepo notice.Repository + TenderRepo tender.TenderRepository +} + +func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository, ollama *ollama.SDK) *NoticeWorker { + return &NoticeWorker{ + Mongo: mongo, + Logger: logger, + Notify: notify, + Ollama: ollama, + NoticeRepo: noticeRepo, + TenderRepo: tenderRepo, + } +} + +func (w *NoticeWorker) Run() { + w.Logger.Info("Notice worker started", map[string]interface{}{}) + + limit := 10 + skip := 0 + for { + notices, _, err := w.NoticeRepo.GetUnProcessedNotices(context.Background(), limit, skip) + if err != nil { + w.Logger.Error("Failed to get notices", map[string]interface{}{ + "error": err.Error(), + }) + continue + } + + if len(notices) == 0 { + break + } + + for _, n := range notices { + w.Logger.Info("Notice", map[string]interface{}{ + "notice": n.ID.Hex(), + }) + + // exists, err := w.TenderRepo.GetByNoticePublicationID(context.Background(), n.NoticePublicationID) + // if err != nil { + // w.Logger.Error("Failed to get tender", map[string]interface{}{ + // "error": err.Error(), + // }) + // continue + // } + + t, err := w.ToTender(&n) + if err != nil { + w.Logger.Error("Failed to create tender", map[string]interface{}{ + "error": err.Error(), + }) + continue + } + + err = w.TenderRepo.Create(context.Background(), t) + if err != nil { + w.Logger.Error("Failed to create tender", map[string]interface{}{ + "error": err.Error(), + }) + } + + } + + skip += limit + } +} + +func (w *NoticeWorker) translateAI(msg string) (string, error) { + builder := w.Ollama.NewChatBuilder().SetStream(false).SetMaxTokens(200000) + chat, err := builder.AddMessage(ollama.MessageRoleUser, fmt.Sprintf("Translate the following text into fluent, natural English while keeping the original meaning and tone. Do not summarize, shorten, or add extra explanations. Just provide the translation without other additional values. %s", msg)).Build() + if err != nil { + return "", err + } + resp, err := w.Ollama.Chat(context.Background(), chat) + if err != nil { + return "", err + } + + return resp.Message.Content, nil +} + +func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) { + // Buyer Organization + buyerOrg := &tender.Organization{} + if n.BuyerOrganization != nil { + buyerOrg = &tender.Organization{ + Name: n.BuyerOrganization.Name, + CompanyID: n.BuyerOrganization.CompanyID, + WebsiteURI: n.BuyerOrganization.WebsiteURI, + ContactName: n.BuyerOrganization.ContactName, + ContactTelephone: n.BuyerOrganization.ContactTelephone, + ContactEmail: n.BuyerOrganization.ContactEmail, + ContactFax: n.BuyerOrganization.ContactFax, + Role: n.BuyerOrganization.Role, + Address: tender.Address{ + StreetName: n.BuyerOrganization.Address.StreetName, + CityName: n.BuyerOrganization.Address.CityName, + PostalZone: n.BuyerOrganization.Address.PostalZone, + CountrySubentityCode: n.BuyerOrganization.Address.CountrySubentityCode, + Department: n.BuyerOrganization.Address.Department, + Region: n.BuyerOrganization.Address.Region, + CountryCode: n.BuyerOrganization.Address.CountryCode, + }, + } + } + + // Review Organization + reviewOrg := &tender.Organization{} + if n.ReviewOrganization != nil { + reviewOrg = &tender.Organization{ + Name: n.ReviewOrganization.Name, + CompanyID: n.ReviewOrganization.CompanyID, + WebsiteURI: n.ReviewOrganization.WebsiteURI, + ContactName: n.ReviewOrganization.ContactName, + ContactTelephone: n.ReviewOrganization.ContactTelephone, + ContactEmail: n.ReviewOrganization.ContactEmail, + ContactFax: n.ReviewOrganization.ContactFax, + Role: n.ReviewOrganization.Role, + Address: tender.Address{ + StreetName: n.ReviewOrganization.Address.StreetName, + CityName: n.ReviewOrganization.Address.CityName, + PostalZone: n.ReviewOrganization.Address.PostalZone, + CountrySubentityCode: n.ReviewOrganization.Address.CountrySubentityCode, + Department: n.ReviewOrganization.Address.Department, + Region: n.ReviewOrganization.Address.Region, + CountryCode: n.ReviewOrganization.Address.CountryCode, + }, + } + } + + // Organizations + organizations := make([]tender.Organization, len(n.Organizations)) + for i, org := range n.Organizations { + organizations[i] = tender.Organization{ + Name: org.Name, + CompanyID: org.CompanyID, + WebsiteURI: org.WebsiteURI, + ContactName: org.ContactName, + ContactTelephone: org.ContactTelephone, + ContactEmail: org.ContactEmail, + ContactFax: org.ContactFax, + Role: org.Role, + Address: tender.Address{ + StreetName: org.Address.StreetName, + CityName: org.Address.CityName, + PostalZone: org.Address.PostalZone, + CountrySubentityCode: org.Address.CountrySubentityCode, + Department: org.Address.Department, + Region: org.Address.Region, + CountryCode: org.Address.CountryCode, + }, + } + } + + // Selection Criteria + selectionCriteria := make([]tender.SelectionCriterion, len(n.SelectionCriteria)) + for i, criterion := range n.SelectionCriteria { + selectionCriteria[i] = tender.SelectionCriterion{ + TypeCode: criterion.TypeCode, + Description: criterion.Description, + LanguageID: criterion.LanguageID, + } + } + + // Winning Tenderer + winningTenderer := &tender.Organization{} + if n.WinningTenderer != nil { + winningTenderer = &tender.Organization{ + Name: n.WinningTenderer.Name, + CompanyID: n.WinningTenderer.CompanyID, + WebsiteURI: n.WinningTenderer.WebsiteURI, + ContactName: n.WinningTenderer.ContactName, + ContactTelephone: n.WinningTenderer.ContactTelephone, + ContactEmail: n.WinningTenderer.ContactEmail, + ContactFax: n.WinningTenderer.ContactFax, + Role: n.WinningTenderer.Role, + Address: tender.Address{ + StreetName: n.WinningTenderer.Address.StreetName, + CityName: n.WinningTenderer.Address.CityName, + PostalZone: n.WinningTenderer.Address.PostalZone, + CountrySubentityCode: n.WinningTenderer.Address.CountrySubentityCode, + Department: n.WinningTenderer.Address.Department, + Region: n.WinningTenderer.Address.Region, + CountryCode: n.WinningTenderer.Address.CountryCode, + }, + } + } + + // Modifications + modifications := make([]tender.TenderModification, len(n.Modifications)) + for i, modification := range n.Modifications { + modifications[i] = tender.TenderModification{ + ModificationDate: modification.ModificationDate, + ModificationReason: modification.ModificationReason, + Description: modification.Description, + LanguageID: modification.LanguageID, + } + } + + // Awarded Entities + awardedEntities := make([]tender.Awarded, len(n.AwardedEntities)) + for i, awarded := range n.AwardedEntities { + awardedEntities[i] = tender.Awarded{ + Name: awarded.Name, + Address: awarded.Address, + Country: awarded.Country, + Amount: awarded.Amount, + Currency: awarded.Currency, + Share: awarded.Share, + AwardDate: awarded.AwardDate, + ContractID: awarded.ContractID, + TenderID: awarded.TenderID, + LotID: awarded.LotID, + CompanyID: awarded.CompanyID, + OrganizationID: awarded.OrganizationID, + } + } + + title, err := w.translateAI(n.Title) + if err != nil { + return nil, fmt.Errorf("failed to translate title: %w", err) + } + + description, err := w.translateAI(n.Description) + if err != nil { + return nil, fmt.Errorf("failed to translate description: %w", err) + } + + return &tender.Tender{ + NoticePublicationID: n.NoticePublicationID, + Title: title, + Description: description, + ProcurementTypeCode: n.ProcurementTypeCode, + ProcedureCode: n.ProcedureCode, + EstimatedValue: n.EstimatedValue, + Currency: n.Currency, + TenderDeadline: n.TenderDeadline, + SubmissionURL: n.SubmissionURL, + BuyerOrganization: buyerOrg, + ReviewOrganization: reviewOrg, + Organizations: organizations, + SelectionCriteria: selectionCriteria, + Status: tender.TenderStatus(n.Status), + Source: tender.TenderSource(n.Source), + SourceFileURL: n.SourceFileURL, + SourceFileName: n.SourceFileName, + TenderID: n.TenderID, + ContractNoticeID: n.ContractNoticeID, + ContractFolderID: n.ContractFolderID, + NoticeTypeCode: n.NoticeTypeCode, + NoticeSubTypeCode: n.NoticeSubTypeCode, + NoticeLanguageCode: n.NoticeLanguageCode, + IssueDate: n.IssueDate, + IssueTime: n.IssueTime, + PublicationDate: n.PublicationDate, + SubmissionDeadline: n.SubmissionDeadline, + ApplicationDeadline: n.ApplicationDeadline, + GazetteID: n.GazetteID, + Duration: n.Duration, + DurationUnit: n.DurationUnit, + PlaceOfPerformance: n.PlaceOfPerformance, + CountryCode: n.CountryCode, + RegionCode: n.RegionCode, + CityName: n.CityName, + PostalCode: n.PostalCode, + DocumentURI: n.DocumentURI, + TenderURL: n.TenderURL, + MainClassification: n.MainClassification, + AdditionalClassifications: n.AdditionalClassifications, + OfficialLanguages: n.OfficialLanguages, + CancellationReason: n.CancellationReason, + CancellationDate: n.CancellationDate, + AwardDate: n.AwardDate, + AwardedValue: n.AwardedValue, + ContractNumber: n.ContractNumber, + SuspensionReason: n.SuspensionReason, + SuspensionDate: n.SuspensionDate, + WinningTenderer: winningTenderer, + Modifications: modifications, + AwardedEntities: awardedEntities, + }, nil +} diff --git a/internal/notice/repository.go b/internal/notice/repository.go index ac235ad..bcf9551 100644 --- a/internal/notice/repository.go +++ b/internal/notice/repository.go @@ -15,6 +15,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) Delete(ctx context.Context, id string) error } @@ -121,6 +122,27 @@ func (r *noticeRepository) GetByContractNoticeID(ctx context.Context, contractNo return &result.Items[0], nil } +// 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} + + pagination := orm.Pagination{ + Limit: limit, + Skip: skip, + } + + result, err := r.ormRepo.FindAll(ctx, filter, pagination) + if err != nil { + r.logger.Error("Failed to get unprocessed notices", map[string]interface{}{ + "error": err.Error(), + }) + return nil, 0, err + } + + return result.Items, result.TotalCount, nil +} + // Update updates an existing notice func (r *noticeRepository) Update(ctx context.Context, notice *Notice) error { notice.UpdatedAt = time.Now().Unix() diff --git a/internal/tender/service.go b/internal/tender/service.go index 77faef6..e21f4ea 100644 --- a/internal/tender/service.go +++ b/internal/tender/service.go @@ -6,7 +6,6 @@ import ( "time" "tm/internal/company" "tm/pkg/logger" - "tm/pkg/ollama" "tm/pkg/response" ) @@ -28,16 +27,14 @@ type Service interface { type tenderService struct { repository TenderRepository companyService company.Service - ollamaSDK ollama.SDK logger logger.Logger } // NewService creates a new tender service -func NewService(repository TenderRepository, companyService company.Service, ollamaSDK ollama.SDK, logger logger.Logger) Service { +func NewService(repository TenderRepository, companyService company.Service, logger logger.Logger) Service { return &tenderService{ repository: repository, companyService: companyService, - ollamaSDK: ollamaSDK, logger: logger, } }