Mazyar
582f8b5c02
continuous-integration/drone/push Build is passing
- Introduced a mutex to ensure only one TED scraper run executes at a time, preventing concurrent executions during startup and scheduled runs. - Implemented a mechanism to check if today's TED scrape has already been completed during startup, logging appropriate messages for both completed and new runs. - Added startup catch-up logic for tender translations and unprocessed notices, ensuring that any missed tasks are executed without blocking the application startup. This update improves the reliability and efficiency of the TED scraper and worker processes, ensuring that all necessary tasks are completed after a server restart.
375 lines
12 KiB
Go
375 lines
12 KiB
Go
package bootstrap
|
|
|
|
import (
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
"tm/cmd/worker/workers"
|
|
"tm/internal/notice"
|
|
notificationDomain "tm/internal/notification"
|
|
"tm/internal/tender"
|
|
ai_summarizer "tm/pkg/ai_summarizer"
|
|
"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, aiClient *ai_summarizer.Client, aiStorage *ai_summarizer.StorageClient) *schedule.CronScheduler {
|
|
// Debug: Log worker config
|
|
appLogger.Info("Worker configuration", map[string]interface{}{
|
|
"worker_interval": config.Worker.Interval,
|
|
"notice_concurrency": config.Worker.NoticeWorkerConcurrency,
|
|
"notice_fetch_batch": config.Worker.NoticeFetchBatchSize,
|
|
"notice_processing_limit": config.Worker.NoticeProcessingLimit,
|
|
"notice_batch_pause": config.Worker.NoticeBatchPause.String(),
|
|
"notice_fetch_error_backoff": config.Worker.NoticeFetchErrorBackoff.String(),
|
|
"translation_enabled": config.Worker.TranslationEnabled,
|
|
"translation_interval": config.Worker.TranslationInterval,
|
|
"notification_interval": config.Worker.NotificationInterval,
|
|
})
|
|
// Initialize repositories
|
|
noticeRepo := notice.NewRepository(mongoManager, appLogger)
|
|
tenderRepo := tender.NewRepository(mongoManager, appLogger)
|
|
notificationRepo := notificationDomain.NewRepository(mongoManager, appLogger)
|
|
|
|
// Create a single shared cron scheduler for all recurring jobs
|
|
scheduler := schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo)
|
|
|
|
// Schedule Notice Worker job (no blocking run at startup; first tick runs on schedule)
|
|
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,
|
|
config.Worker.NoticeProcessingLimit,
|
|
config.Worker.NoticeWorkerConcurrency,
|
|
config.Worker.NoticeFetchBatchSize,
|
|
config.Worker.DeleteProcessedNotices,
|
|
config.Worker.NoticeBatchPause,
|
|
config.Worker.NoticeFetchErrorBackoff,
|
|
)
|
|
worker.Run()
|
|
},
|
|
Expr: workerInterval,
|
|
})
|
|
|
|
// Initialize translation backfill worker for configured languages
|
|
if !config.Worker.TranslationEnabled {
|
|
appLogger.Info("Tender translation worker disabled by configuration", map[string]interface{}{
|
|
"translation_enabled": false,
|
|
})
|
|
} else if aiClient != nil {
|
|
targetLanguages := parseTranslationLanguages(config.AISummarizer)
|
|
translationSuccessCounter := mongo.NewCounter(mongoManager)
|
|
scheduler.AddJob(schedule.Job{
|
|
Name: "Tender Translation Worker Job",
|
|
Func: func() {
|
|
worker := workers.NewTranslationWorker(
|
|
mongoManager,
|
|
appLogger,
|
|
tenderRepo,
|
|
aiClient,
|
|
aiStorage,
|
|
translationSuccessCounter,
|
|
targetLanguages,
|
|
config.Worker.TranslationBatchSize,
|
|
)
|
|
worker.Run()
|
|
},
|
|
Expr: config.Worker.TranslationInterval,
|
|
})
|
|
appLogger.Info("Scheduled tender translation worker", map[string]interface{}{
|
|
"target_languages": targetLanguages,
|
|
"interval": config.Worker.TranslationInterval,
|
|
"batch_size": config.Worker.TranslationBatchSize,
|
|
})
|
|
|
|
go func() {
|
|
appLogger.Info("Running startup catch-up for tender translations", map[string]interface{}{
|
|
"target_languages": targetLanguages,
|
|
})
|
|
worker := workers.NewTranslationWorker(
|
|
mongoManager,
|
|
appLogger,
|
|
tenderRepo,
|
|
aiClient,
|
|
aiStorage,
|
|
translationSuccessCounter,
|
|
targetLanguages,
|
|
config.Worker.TranslationBatchSize,
|
|
)
|
|
worker.Run()
|
|
}()
|
|
} else {
|
|
appLogger.Warn("AI summarizer client not available, tender translation worker is disabled", map[string]interface{}{})
|
|
}
|
|
|
|
notificationInterval := config.Worker.NotificationInterval
|
|
if notificationInterval == "" {
|
|
notificationInterval = "0 * * * * *"
|
|
appLogger.Warn("WORKER_NOTIFICATION_INTERVAL not set, using default schedule", map[string]interface{}{
|
|
"interval": notificationInterval,
|
|
})
|
|
}
|
|
|
|
scheduler.AddJob(schedule.Job{
|
|
Name: "Scheduled Notification Worker Job",
|
|
Func: func() {
|
|
worker := workers.NewNotificationWorker(notificationRepo, appLogger)
|
|
worker.Run()
|
|
},
|
|
Expr: notificationInterval,
|
|
})
|
|
appLogger.Info("Scheduled notification delivery worker", map[string]interface{}{
|
|
"interval": notificationInterval,
|
|
})
|
|
|
|
// After a server restart, process any unprocessed notices (including today's) without blocking startup.
|
|
go func() {
|
|
appLogger.Info("Running startup catch-up for unprocessed notices", map[string]interface{}{})
|
|
w := workers.NewNoticeWorker(
|
|
mongoManager,
|
|
appLogger,
|
|
¬ify,
|
|
noticeRepo,
|
|
tenderRepo,
|
|
config.Worker.NoticeProcessingLimit,
|
|
config.Worker.NoticeWorkerConcurrency,
|
|
config.Worker.NoticeFetchBatchSize,
|
|
config.Worker.DeleteProcessedNotices,
|
|
config.Worker.NoticeBatchPause,
|
|
config.Worker.NoticeFetchErrorBackoff,
|
|
)
|
|
w.Run()
|
|
}()
|
|
|
|
// 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, mongoManager *mongo.ConnectionManager, 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
|
|
}
|
|
|
|
translationCounter := mongo.NewCounter(mongoManager)
|
|
cfg := &ai_summarizer.Config{
|
|
APIBaseURL: conf.APIBaseURL,
|
|
APITimeout: conf.APITimeout,
|
|
APIRetryCount: conf.APIRetryCount,
|
|
APIRetryDelay: conf.APIRetryDelay,
|
|
OnSuccessfulTranslation: mongo.AITranslationSuccessCallback(translationCounter),
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
// InitAISummarizerStorage initializes the AI summarizer MinIO storage client for workers.
|
|
func InitAISummarizerStorage(conf AISummarizerConfig, log logger.Logger) *ai_summarizer.StorageClient {
|
|
if conf.MinioEndpoint == "" {
|
|
log.Warn("AI Summarizer MinIO endpoint not configured, translation sync from storage will be unavailable", map[string]interface{}{})
|
|
return nil
|
|
}
|
|
|
|
cfg := &ai_summarizer.Config{
|
|
APIBaseURL: conf.APIBaseURL,
|
|
APITimeout: conf.APITimeout,
|
|
APIRetryCount: conf.APIRetryCount,
|
|
APIRetryDelay: conf.APIRetryDelay,
|
|
MinioEndpoint: conf.MinioEndpoint,
|
|
MinioAccessKey: conf.MinioAccessKey,
|
|
MinioSecretKey: conf.MinioSecretKey,
|
|
MinioUseSSL: conf.MinioUseSSL,
|
|
MinioRegion: conf.MinioRegion,
|
|
MinioBucket: conf.MinioBucket,
|
|
}
|
|
|
|
storage, err := ai_summarizer.NewStorageClient(cfg, log)
|
|
if err != nil {
|
|
log.Error("Failed to initialize AI Summarizer storage client for worker", map[string]interface{}{
|
|
"error": err.Error(),
|
|
})
|
|
return nil
|
|
}
|
|
|
|
log.Info("AI Summarizer storage client initialized for worker", map[string]interface{}{
|
|
"endpoint": conf.MinioEndpoint,
|
|
"bucket": conf.MinioBucket,
|
|
})
|
|
|
|
return storage
|
|
}
|
|
|
|
func parseTranslationLanguages(conf AISummarizerConfig) []string {
|
|
raw := strings.TrimSpace(conf.TranslationLanguages)
|
|
if raw == "" {
|
|
raw = conf.DefaultLanguage
|
|
}
|
|
parts := strings.Split(raw, ",")
|
|
langs := make([]string, 0, len(parts))
|
|
seen := make(map[string]struct{}, len(parts))
|
|
for _, part := range parts {
|
|
lang := strings.ToLower(strings.TrimSpace(part))
|
|
if lang == "" {
|
|
continue
|
|
}
|
|
if _, dup := seen[lang]; dup {
|
|
continue
|
|
}
|
|
seen[lang] = struct{}{}
|
|
langs = append(langs, lang)
|
|
}
|
|
if len(langs) == 0 {
|
|
return []string{"en"}
|
|
}
|
|
return langs
|
|
}
|
|
|
|
// 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
|
|
}
|