Mazyar
1ad0206e61
continuous-integration/drone/push Build is passing
- Renamed and refactored the AI pipeline auto run functionality to a daily run, enhancing clarity and purpose. - Introduced a new `AIPipelineDailyWorker` to manage the daily execution of the AI pipeline, replacing the previous auto run implementation. - Updated configuration fields and logging messages to reflect the change from auto to daily run, ensuring consistent terminology throughout the codebase. - Removed the obsolete `ai_pipeline_auto.go` file to streamline the worker structure. This update improves the maintainability and readability of the AI pipeline management by clearly distinguishing between auto and daily run functionalities.
469 lines
15 KiB
Go
469 lines
15 KiB
Go
package bootstrap
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"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"
|
|
)
|
|
|
|
// aiPipelineDailyRunMu ensures only one AI pipeline daily-run executes at a time (startup catch-up and cron).
|
|
var aiPipelineDailyRunMu 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_daily_enabled": config.Worker.AIPipelineAutoEnabled,
|
|
"ai_pipeline_daily_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 daily-run worker disabled by configuration", map[string]interface{}{
|
|
"ai_pipeline_daily_enabled": false,
|
|
})
|
|
} else if aiClient != nil {
|
|
dailyJobTracker := schedule.NewDailyJobTracker(mongoManager, appLogger)
|
|
|
|
runAIPipelineDaily := func(trigger string) {
|
|
aiPipelineDailyRunMu.Lock()
|
|
defer aiPipelineDailyRunMu.Unlock()
|
|
|
|
ctx := context.Background()
|
|
today := time.Now().Local()
|
|
|
|
completed, checkErr := dailyJobTracker.IsCompleted(ctx, schedule.AIPipelineDailyJobName, today)
|
|
if checkErr != nil {
|
|
appLogger.Error("Failed to check AI pipeline daily-run completion status", map[string]interface{}{
|
|
"trigger": trigger,
|
|
"error": checkErr.Error(),
|
|
"date": today.Format("02/01/2006"),
|
|
})
|
|
} else if completed {
|
|
appLogger.Info("AI pipeline daily-run 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 daily-run", map[string]interface{}{
|
|
"date": today.Format("02/01/2006"),
|
|
})
|
|
} else {
|
|
appLogger.Info("Running scheduled AI pipeline daily-run", map[string]interface{}{
|
|
"date": today.Format("02/01/2006"),
|
|
})
|
|
}
|
|
|
|
worker := workers.NewAIPipelineDailyWorker(appLogger, aiClient)
|
|
if err := worker.Run(); err != nil {
|
|
if errors.Is(err, ai_summarizer.ErrPipelineJobRunning) {
|
|
appLogger.Info("AI pipeline daily-run deferred: job already running", map[string]interface{}{
|
|
"trigger": trigger,
|
|
"date": today.Format("02/01/2006"),
|
|
})
|
|
} else {
|
|
appLogger.Error("AI pipeline daily-run failed", map[string]interface{}{
|
|
"trigger": trigger,
|
|
"date": today.Format("02/01/2006"),
|
|
"error": err.Error(),
|
|
})
|
|
}
|
|
return
|
|
}
|
|
|
|
if markErr := dailyJobTracker.MarkCompleted(ctx, schedule.AIPipelineDailyJobName, today); markErr != nil {
|
|
appLogger.Error("Failed to mark AI pipeline daily-run as completed", map[string]interface{}{
|
|
"trigger": trigger,
|
|
"date": today.Format("02/01/2006"),
|
|
"error": markErr.Error(),
|
|
})
|
|
}
|
|
|
|
appLogger.Info("AI pipeline daily-run completed successfully", map[string]interface{}{
|
|
"trigger": trigger,
|
|
"date": today.Format("02/01/2006"),
|
|
})
|
|
}
|
|
|
|
scheduler.AddJob(schedule.Job{
|
|
Name: "AI Pipeline Daily Run Job",
|
|
Func: func() { runAIPipelineDaily("scheduled") },
|
|
Expr: config.Worker.AIPipelineAutoInterval,
|
|
})
|
|
appLogger.Info("Scheduled AI pipeline daily-run worker", map[string]interface{}{
|
|
"interval": config.Worker.AIPipelineAutoInterval,
|
|
})
|
|
|
|
go func() {
|
|
runAIPipelineDaily("startup")
|
|
}()
|
|
} else {
|
|
appLogger.Warn("AI summarizer client not available, AI pipeline daily-run 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
|
|
}
|