Mazyar
492f9ba3c8
continuous-integration/drone/push Build is passing
- Introduced AIPipelineAutoWorker to manage the execution of the AI pipeline auto run, including startup catch-up and scheduled tasks. - Enhanced WorkerConfig to include AIPipelineAutoEnabled and AIPipelineAutoInterval settings for better control over AI pipeline execution. - Added logging for AI pipeline auto run status, including success and error handling, to improve observability. - Updated daily job tracker to include AIPipelineAutoJobName for tracking AI pipeline job completions. This update enhances the system's capability to automate AI pipeline executions, improving efficiency and reliability in processing AI tasks.
461 lines
14 KiB
Go
461 lines
14 KiB
Go
package bootstrap
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"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"
|
|
)
|
|
|
|
// aiPipelineAutoRunMu ensures only one AI pipeline auto run executes at a time (startup catch-up and cron).
|
|
var aiPipelineAutoRunMu sync.Mutex
|
|
|
|
// 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,
|
|
"ai_pipeline_auto_enabled": config.Worker.AIPipelineAutoEnabled,
|
|
"ai_pipeline_auto_interval": config.Worker.AIPipelineAutoInterval,
|
|
})
|
|
// 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,
|
|
})
|
|
|
|
if !config.Worker.AIPipelineAutoEnabled {
|
|
appLogger.Info("AI pipeline auto worker disabled by configuration", map[string]interface{}{
|
|
"ai_pipeline_auto_enabled": false,
|
|
})
|
|
} else if aiClient != nil {
|
|
dailyJobTracker := schedule.NewDailyJobTracker(mongoManager, appLogger)
|
|
|
|
runAIPipelineAuto := func(trigger string) {
|
|
aiPipelineAutoRunMu.Lock()
|
|
defer aiPipelineAutoRunMu.Unlock()
|
|
|
|
ctx := context.Background()
|
|
today := time.Now().Local()
|
|
|
|
completed, checkErr := dailyJobTracker.IsCompleted(ctx, schedule.AIPipelineAutoJobName, today)
|
|
if checkErr != nil {
|
|
appLogger.Error("Failed to check AI pipeline auto completion status", map[string]interface{}{
|
|
"trigger": trigger,
|
|
"error": checkErr.Error(),
|
|
"date": today.Format("02/01/2006"),
|
|
})
|
|
} else if completed {
|
|
appLogger.Info("AI pipeline auto skipped: already completed today", map[string]interface{}{
|
|
"trigger": trigger,
|
|
"date": today.Format("02/01/2006"),
|
|
})
|
|
return
|
|
}
|
|
|
|
if trigger == "startup" {
|
|
appLogger.Info("Running startup catch-up for today's AI pipeline auto", map[string]interface{}{
|
|
"date": today.Format("02/01/2006"),
|
|
})
|
|
} else {
|
|
appLogger.Info("Running scheduled AI pipeline auto", map[string]interface{}{
|
|
"date": today.Format("02/01/2006"),
|
|
})
|
|
}
|
|
|
|
worker := workers.NewAIPipelineAutoWorker(appLogger, aiClient)
|
|
if err := worker.Run(); err != nil {
|
|
appLogger.Error("AI pipeline auto run failed", map[string]interface{}{
|
|
"trigger": trigger,
|
|
"date": today.Format("02/01/2006"),
|
|
"error": err.Error(),
|
|
})
|
|
return
|
|
}
|
|
|
|
if markErr := dailyJobTracker.MarkCompleted(ctx, schedule.AIPipelineAutoJobName, today); markErr != nil {
|
|
appLogger.Error("Failed to mark AI pipeline auto as completed", map[string]interface{}{
|
|
"trigger": trigger,
|
|
"date": today.Format("02/01/2006"),
|
|
"error": markErr.Error(),
|
|
})
|
|
}
|
|
|
|
appLogger.Info("AI pipeline auto run completed successfully", map[string]interface{}{
|
|
"trigger": trigger,
|
|
"date": today.Format("02/01/2006"),
|
|
})
|
|
}
|
|
|
|
scheduler.AddJob(schedule.Job{
|
|
Name: "AI Pipeline Auto Job",
|
|
Func: func() { runAIPipelineAuto("scheduled") },
|
|
Expr: config.Worker.AIPipelineAutoInterval,
|
|
})
|
|
appLogger.Info("Scheduled AI pipeline auto worker", map[string]interface{}{
|
|
"interval": config.Worker.AIPipelineAutoInterval,
|
|
})
|
|
|
|
go func() {
|
|
runAIPipelineAuto("startup")
|
|
}()
|
|
} else {
|
|
appLogger.Warn("AI summarizer client not available, AI pipeline auto worker is disabled", map[string]interface{}{})
|
|
}
|
|
|
|
// 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
|
|
}
|