Mazyar
ca35eb5f15
continuous-integration/drone/push Build is passing
- Added `TranslationEnabled` and `TranslationInterval` fields to the worker configuration to manage automatic translation scheduling. - Updated the worker initialization to log when the translation worker is disabled. - Improved error handling in the AI summarizer client by introducing `APIStatusError` for better context on API failures, replacing direct error messages with structured error responses. This update enhances the configurability of the worker and improves error reporting for AI service interactions, contributing to better maintainability and user experience.
348 lines
11 KiB
Go
348 lines
11 KiB
Go
package bootstrap
|
|
|
|
import (
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
"tm/cmd/worker/workers"
|
|
"tm/internal/notice"
|
|
"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,
|
|
})
|
|
// 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)
|
|
|
|
// Document summarization via external AI service (requires AI_SUMMARIZER_API_BASE_URL)
|
|
if aiClient != nil {
|
|
scheduler.AddJob(schedule.Job{
|
|
Name: "Document Summarization Worker Job",
|
|
Func: func() {
|
|
worker := workers.NewDocumentSummarizationWorker(mongoManager, appLogger, tenderRepo, aiClient, aiStorage)
|
|
worker.Run()
|
|
},
|
|
Expr: "0 11 * * * *", // Run at 11 AM every day (after AI pipeline scrape)
|
|
})
|
|
} else {
|
|
appLogger.Warn("Document summarization worker not scheduled: AI summarizer API is not configured", map[string]interface{}{})
|
|
}
|
|
|
|
// 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,
|
|
})
|
|
} else {
|
|
appLogger.Warn("AI summarizer client not available, tender translation worker is disabled", map[string]interface{}{})
|
|
}
|
|
|
|
// Kick off one notice-processing pass without blocking startup (cron continues on schedule)
|
|
go func() {
|
|
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
|
|
}
|