Mazyar
405 lines
13 KiB
Go
405 lines
13 KiB
Go
package bootstrap
|
|
|
|
import (
|
|
"fmt"
|
|
"time"
|
|
"tm/cmd/worker/workers"
|
|
"tm/internal/notice"
|
|
"tm/internal/tender"
|
|
ai_summarizer "tm/pkg/ai_summarizer"
|
|
"tm/pkg/config"
|
|
"tm/pkg/glm"
|
|
"tm/pkg/logger"
|
|
"tm/pkg/minio"
|
|
"tm/pkg/mongo"
|
|
"tm/pkg/notification"
|
|
"tm/pkg/ollama"
|
|
"tm/pkg/queue"
|
|
"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, glmService *glm.SDK, minioService *minio.Service, aiClient *ai_summarizer.Client) *schedule.CronScheduler {
|
|
// Debug: Log worker config
|
|
appLogger.Info("Worker configuration", map[string]interface{}{
|
|
"worker_interval": config.Worker.Interval,
|
|
"queue_enabled": config.Queue.Enabled,
|
|
"queue_mode": config.Queue.Mode,
|
|
})
|
|
// Initialize repositories
|
|
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,
|
|
})
|
|
noticeWorker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService, config.Worker.NoticeProcessingLimit, config.Worker.DeleteProcessedNotices)
|
|
noticeWorker.Run()
|
|
|
|
// Initialize Queue-based scraping if enabled
|
|
var queueService queue.Queue
|
|
if config.Queue.Enabled {
|
|
var err error
|
|
queueConfig := queue.QueueConfig{
|
|
RedisURL: config.Queue.RedisURL,
|
|
MaxRetries: config.Queue.MaxRetries,
|
|
RetryBackoff: config.Queue.RetryBackoff,
|
|
JobTimeout: config.Queue.JobTimeout,
|
|
PoolSize: config.Queue.PoolSize,
|
|
IsDLQEnabled: config.Queue.IsDLQEnabled,
|
|
MaxQueueLength: 0, // Unlimited
|
|
}
|
|
|
|
queueService, err = queue.NewRedisQueue(queueConfig, appLogger)
|
|
if err != nil {
|
|
appLogger.Error("Failed to initialize queue service", map[string]interface{}{
|
|
"error": err.Error(),
|
|
})
|
|
appLogger.Warn("Queue-based scraping is disabled, falling back to scheduled scraping", map[string]interface{}{})
|
|
config.Queue.Enabled = false
|
|
} else {
|
|
appLogger.Info("Queue service initialized successfully", map[string]interface{}{
|
|
"mode": config.Queue.Mode,
|
|
})
|
|
}
|
|
}
|
|
|
|
// Start Queue Producer Worker (enqueues unscraped tenders periodically)
|
|
if config.Queue.Enabled && (config.Queue.Mode == "producer" || config.Queue.Mode == "both") {
|
|
appLogger.Info("Scheduling queue producer worker", map[string]interface{}{
|
|
"batch_size": config.Queue.ProducerBatch,
|
|
"interval": config.Queue.ProducerInterval,
|
|
})
|
|
|
|
scheduler.AddJob(schedule.Job{
|
|
Name: "Queue Producer Worker Job",
|
|
Func: func() {
|
|
producer := workers.NewQueueProducerWorker(
|
|
mongoManager,
|
|
appLogger,
|
|
tenderRepo,
|
|
queueService,
|
|
config.Queue.ProducerBatch,
|
|
)
|
|
producer.Run()
|
|
},
|
|
Expr: config.Queue.ProducerInterval,
|
|
})
|
|
}
|
|
|
|
// Initialize Document Summarization Worker (only if MinIO is available)
|
|
if minioService != nil {
|
|
appLogger.Info("Starting document summarization worker", map[string]interface{}{})
|
|
summarizerWorker := workers.NewDocumentSummarizationWorker(mongoManager, appLogger, tenderRepo, minioService, glmService)
|
|
summarizerWorker.Run()
|
|
|
|
// Schedule Document Summarization Worker job
|
|
scheduler.AddJob(schedule.Job{
|
|
Name: "Document Summarization Worker Job",
|
|
Func: func() {
|
|
worker := workers.NewDocumentSummarizationWorker(mongoManager, appLogger, tenderRepo, minioService, glmService)
|
|
worker.Run()
|
|
},
|
|
Expr: "0 11 * * * *", // Run at 11 AM every day (1 hour after document scraping)
|
|
})
|
|
} else {
|
|
appLogger.Warn("MinIO service not available, document summarization will be skipped", map[string]interface{}{})
|
|
}
|
|
|
|
// Schedule Notice Worker job
|
|
workerInterval := config.Worker.Interval
|
|
if workerInterval == "" {
|
|
workerInterval = "* 10 * * * *" // Default: every hour at minute 10
|
|
appLogger.Warn("WORKER_INTERVAL not set, using default schedule", map[string]interface{}{
|
|
"interval": workerInterval,
|
|
})
|
|
}
|
|
|
|
scheduler.AddJob(schedule.Job{
|
|
Name: "Notice Worker Job",
|
|
Func: func() {
|
|
worker := workers.NewNoticeWorker(mongoManager, appLogger, ¬ify, noticeRepo, tenderRepo, glmService, config.Worker.NoticeProcessingLimit, config.Worker.DeleteProcessedNotices)
|
|
worker.Run()
|
|
},
|
|
Expr: workerInterval,
|
|
})
|
|
|
|
// Initialize translation backfill worker for default language
|
|
if aiClient != nil {
|
|
targetLanguage := config.AISummarizer.DefaultLanguage
|
|
scheduler.AddJob(schedule.Job{
|
|
Name: "Tender Translation Worker Job",
|
|
Func: func() {
|
|
worker := workers.NewTranslationWorker(
|
|
mongoManager,
|
|
appLogger,
|
|
tenderRepo,
|
|
aiClient,
|
|
targetLanguage,
|
|
config.Worker.TranslationBatchSize,
|
|
)
|
|
worker.Run()
|
|
},
|
|
Expr: config.Worker.TranslationInterval,
|
|
})
|
|
appLogger.Info("Scheduled tender translation worker", map[string]interface{}{
|
|
"target_language": targetLanguage,
|
|
"interval": config.Worker.TranslationInterval,
|
|
"batch_size": config.Worker.TranslationBatchSize,
|
|
})
|
|
} else {
|
|
appLogger.Warn("AI summarizer client not available, tender translation worker is disabled", map[string]interface{}{})
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
// InitAISummarizerClient initializes the AI summarizer HTTP client used by worker jobs.
|
|
func InitAISummarizerClient(conf AISummarizerConfig, log logger.Logger) *ai_summarizer.Client {
|
|
if conf.APIBaseURL == "" {
|
|
log.Warn("AI Summarizer API base URL not configured, translation worker will be unavailable", map[string]interface{}{})
|
|
return nil
|
|
}
|
|
|
|
cfg := &ai_summarizer.Config{
|
|
APIBaseURL: conf.APIBaseURL,
|
|
APITimeout: conf.APITimeout,
|
|
APIRetryCount: conf.APIRetryCount,
|
|
APIRetryDelay: conf.APIRetryDelay,
|
|
}
|
|
|
|
client, err := ai_summarizer.NewClient(cfg, log)
|
|
if err != nil {
|
|
log.Error("Failed to initialize AI Summarizer client for worker", map[string]interface{}{
|
|
"error": err.Error(),
|
|
})
|
|
return nil
|
|
}
|
|
|
|
log.Info("AI Summarizer client initialized for worker", map[string]interface{}{
|
|
"base_url": conf.APIBaseURL,
|
|
"timeout": conf.APITimeout,
|
|
})
|
|
|
|
return client
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
// Init GLM Service
|
|
func InitGLMService(conf GLMConfig, log logger.Logger) *glm.SDK {
|
|
glmSDK, err := glm.New(
|
|
&glm.Config{
|
|
BaseURL: conf.BaseURL,
|
|
APIKey: conf.APIKey,
|
|
Timeout: conf.Timeout,
|
|
RetryAttempts: conf.RetryAttempts,
|
|
RetryDelay: conf.RetryDelay,
|
|
EnableLogging: conf.EnableLogging,
|
|
UserAgent: conf.UserAgent,
|
|
DefaultModel: conf.DefaultModel,
|
|
DefaultMaxTokens: conf.DefaultMaxTokens,
|
|
DefaultTopP: conf.DefaultTopP,
|
|
DefaultTemperature: conf.DefaultTemperature,
|
|
DefaultFrequencyPenalty: conf.DefaultFrequencyPenalty,
|
|
DefaultPresencePenalty: conf.DefaultPresencePenalty,
|
|
EnableStreaming: conf.EnableStreaming,
|
|
EnableThinking: conf.EnableThinking,
|
|
TranslationOptions: glm.TranslationOption{
|
|
Model: conf.TranslateOptions.Model,
|
|
MaxTokens: conf.TranslateOptions.MaxTokens,
|
|
Temperature: conf.TranslateOptions.Temperature,
|
|
FrequencyPenalty: conf.TranslateOptions.FrequencyPenalty,
|
|
PresencePenalty: conf.TranslateOptions.PresencePenalty,
|
|
TopP: conf.TranslateOptions.TopP,
|
|
EnableThinking: conf.TranslateOptions.EnableThinking,
|
|
},
|
|
}, log)
|
|
|
|
if err != nil {
|
|
log.Error("Failed to initialize GLM SDK", map[string]interface{}{
|
|
"error": err.Error(),
|
|
})
|
|
log.Warn("GLM service will be disabled, translation features will not be available", map[string]interface{}{})
|
|
return nil
|
|
}
|
|
|
|
log.Info("GLM SDK initialized successfully", map[string]interface{}{
|
|
"base_url": conf.BaseURL,
|
|
"timeout": conf.Timeout,
|
|
})
|
|
|
|
return glmSDK
|
|
}
|
|
|
|
// Init MinIO Service
|
|
func InitMinIOService(conf config.MinIOConfig, log logger.Logger) *minio.Service {
|
|
// Convert bootstrap config to minio config
|
|
minioConfig := minio.Config{
|
|
Endpoint: conf.Endpoint,
|
|
AccessKeyID: conf.AccessKeyID,
|
|
SecretAccessKey: conf.SecretAccessKey,
|
|
UseSSL: conf.UseSSL,
|
|
Region: conf.Region,
|
|
DefaultBucket: conf.DefaultBucket,
|
|
HierarchicalBucket: conf.HierarchicalBucket,
|
|
}
|
|
|
|
// Validate config
|
|
if err := minioConfig.Validate(); err != nil {
|
|
log.Warn("Invalid MinIO configuration, MinIO features will be disabled", map[string]interface{}{
|
|
"error": err.Error(),
|
|
})
|
|
return nil // Return nil instead of panicking
|
|
}
|
|
|
|
// Create connection manager
|
|
connManager, err := minio.NewConnectionManager(minioConfig, log)
|
|
if err != nil {
|
|
log.Error("Failed to create MinIO connection manager", map[string]interface{}{
|
|
"error": err.Error(),
|
|
})
|
|
panic(fmt.Sprintf("Failed to create MinIO connection manager: %v", err))
|
|
}
|
|
|
|
// Connect to MinIO
|
|
if err := connManager.Connect(); err != nil {
|
|
log.Warn("Failed to connect to MinIO, MinIO features will be disabled", map[string]interface{}{
|
|
"endpoint": minioConfig.Endpoint,
|
|
"error": err.Error(),
|
|
})
|
|
return nil // Return nil instead of panicking
|
|
}
|
|
|
|
// Create service
|
|
service := minio.NewService(connManager, minioConfig, log)
|
|
|
|
log.Info("MinIO service initialized successfully", map[string]interface{}{
|
|
"endpoint": conf.Endpoint,
|
|
"default_bucket": conf.DefaultBucket,
|
|
"hierarchical_bucket": conf.HierarchicalBucket,
|
|
})
|
|
|
|
return service
|
|
}
|