diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index 6c70d08..bdd5080 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -23,7 +23,7 @@ import ( "tm/pkg/schedule" ) -// aiPipelineDailyRunMu ensures only one AI pipeline daily-run executes at a time (startup catch-up and cron). +// aiPipelineDailyRunMu ensures only one AI pipeline auto run executes at a time (startup catch-up and cron). var aiPipelineDailyRunMu sync.Mutex // Init Application Configuration @@ -98,8 +98,8 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger "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, + "ai_pipeline_auto_enabled": config.Worker.AIPipelineAutoEnabled, + "ai_pipeline_auto_interval": config.Worker.AIPipelineAutoInterval, }) // Initialize repositories noticeRepo := notice.NewRepository(mongoManager, appLogger) @@ -231,8 +231,8 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger }) if !config.Worker.AIPipelineAutoEnabled { - appLogger.Info("AI pipeline daily-run worker disabled by configuration", map[string]interface{}{ - "ai_pipeline_daily_enabled": false, + 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) @@ -246,13 +246,13 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger 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{}{ + 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 daily-run skipped: already completed today", map[string]interface{}{ + appLogger.Info("AI pipeline auto skipped: already completed today", map[string]interface{}{ "trigger": trigger, "date": today.Format("02/01/2006"), }) @@ -260,11 +260,11 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger } if trigger == "startup" { - appLogger.Info("Running startup catch-up for today's AI pipeline daily-run", map[string]interface{}{ + appLogger.Info("Running startup catch-up for today's AI pipeline auto run", map[string]interface{}{ "date": today.Format("02/01/2006"), }) } else { - appLogger.Info("Running scheduled AI pipeline daily-run", map[string]interface{}{ + appLogger.Info("Running scheduled AI pipeline auto run", map[string]interface{}{ "date": today.Format("02/01/2006"), }) } @@ -272,12 +272,12 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger 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{}{ + appLogger.Info("AI pipeline auto 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{}{ + appLogger.Error("AI pipeline auto failed", map[string]interface{}{ "trigger": trigger, "date": today.Format("02/01/2006"), "error": err.Error(), @@ -287,14 +287,14 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger } if markErr := dailyJobTracker.MarkCompleted(ctx, schedule.AIPipelineDailyJobName, today); markErr != nil { - appLogger.Error("Failed to mark AI pipeline daily-run as completed", map[string]interface{}{ + 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 daily-run completed successfully", map[string]interface{}{ + appLogger.Info("AI pipeline auto completed successfully", map[string]interface{}{ "trigger": trigger, "date": today.Format("02/01/2006"), }) @@ -305,11 +305,11 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger } scheduler.AddJob(schedule.Job{ - Name: "AI Pipeline Daily Run Job", + Name: "AI Pipeline Auto Job", Func: func() { runAIPipelineDaily("scheduled") }, Expr: config.Worker.AIPipelineAutoInterval, }) - appLogger.Info("Scheduled AI pipeline daily-run worker", map[string]interface{}{ + appLogger.Info("Scheduled AI pipeline auto worker", map[string]interface{}{ "interval": config.Worker.AIPipelineAutoInterval, }) @@ -317,7 +317,7 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger runAIPipelineDaily("startup") }() } else { - appLogger.Warn("AI summarizer client not available, AI pipeline daily-run worker is disabled", map[string]interface{}{}) + 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. diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index e8652ed..363cd77 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -36,12 +36,12 @@ type WorkerConfig struct { TranslationEnabled bool `env:"WORKER_TRANSLATION_ENABLED" envDefault:"true"` // NotificationInterval schedules promotion of due scheduled notifications from pending to sent. NotificationInterval string `env:"WORKER_NOTIFICATION_INTERVAL" envDefault:"0 * * * * *"` - // AIPipelineAutoEnabled schedules the daily AI pipeline daily-run on the worker. + // AIPipelineAutoEnabled schedules the daily AI pipeline auto run on the worker. // On-demand pipeline routes on web are unaffected. AIPipelineAutoEnabled bool `env:"WORKER_AI_PIPELINE_AUTO_ENABLED" envDefault:"true"` - // AIPipelineAutoInterval runs daily-run once per day by default at 11:00 UTC (after TED scrape at 10:00). + // AIPipelineAutoInterval runs pipeline auto once per day by default at 11:00 UTC (after TED scrape at 10:00). AIPipelineAutoInterval string `env:"WORKER_AI_PIPELINE_AUTO_INTERVAL" envDefault:"0 0 11 * * *"` - // RecommendationRefreshAfterPipelineEnabled refreshes cached company recommendations after the daily AI pipeline run. + // RecommendationRefreshAfterPipelineEnabled refreshes cached company recommendations after the daily AI pipeline auto run. RecommendationRefreshAfterPipelineEnabled bool `env:"WORKER_RECOMMENDATION_REFRESH" envDefault:"true"` } diff --git a/cmd/worker/workers/ai_pipeline_daily.go b/cmd/worker/workers/ai_pipeline_daily.go index d3684df..627f2f8 100644 --- a/cmd/worker/workers/ai_pipeline_daily.go +++ b/cmd/worker/workers/ai_pipeline_daily.go @@ -8,13 +8,13 @@ import ( "tm/pkg/logger" ) -// AIPipelineDailyWorker triggers the Opplens AI pipeline daily run (sync + full pipeline on newly synced tenders). +// AIPipelineDailyWorker triggers the scheduled Opplens AI pipeline auto run. type AIPipelineDailyWorker struct { Logger logger.Logger AIClient *ai_summarizer.Client } -// NewAIPipelineDailyWorker creates a worker that calls POST /pipeline/daily-run on the AI service. +// NewAIPipelineDailyWorker creates a worker that calls POST /pipeline/auto on the AI service. func NewAIPipelineDailyWorker(log logger.Logger, aiClient *ai_summarizer.Client) *AIPipelineDailyWorker { return &AIPipelineDailyWorker{ Logger: log, @@ -22,28 +22,28 @@ func NewAIPipelineDailyWorker(log logger.Logger, aiClient *ai_summarizer.Client) } } -// Run starts the pipeline daily-run job. A 409 response means a run is already in progress and is returned as an error +// Run starts the pipeline auto job. A 409 response means a run is already in progress and is returned as an error // so the daily job tracker does not mark today as completed before the run actually starts. func (w *AIPipelineDailyWorker) Run() error { if w.AIClient == nil { - w.Logger.Warn("AI pipeline daily worker skipped: AI client is not configured", map[string]interface{}{}) + w.Logger.Warn("AI pipeline auto worker skipped: AI client is not configured", map[string]interface{}{}) return nil } ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) defer cancel() - w.Logger.Info("AI pipeline daily worker started", map[string]interface{}{}) + w.Logger.Info("AI pipeline auto worker started", map[string]interface{}{}) - resp, err := w.AIClient.PipelineDailyRun(ctx) + resp, err := w.AIClient.PipelineAuto(ctx) if err != nil { - w.Logger.Error("AI pipeline daily worker failed", map[string]interface{}{ + w.Logger.Error("AI pipeline auto worker failed", map[string]interface{}{ "error": err.Error(), }) return err } - w.Logger.Info("AI pipeline daily worker triggered successfully", map[string]interface{}{ + w.Logger.Info("AI pipeline auto worker triggered successfully", map[string]interface{}{ "status": resp.Status, "report": resp.Report, }) diff --git a/internal/company/recommendation_pipeline_refresh.go b/internal/company/recommendation_pipeline_refresh.go index b1222c4..49d83a8 100644 --- a/internal/company/recommendation_pipeline_refresh.go +++ b/internal/company/recommendation_pipeline_refresh.go @@ -13,7 +13,7 @@ import ( // AIPipelineStatusClient reports pipeline job status from the AI service. type AIPipelineStatusClient interface { - GetPipelineLastRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error) + GetPipelineLastAutoRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error) } const ( @@ -22,7 +22,7 @@ const ( recommendationRefreshConcurrency = 3 ) -// ScheduleRefreshCachedAIRecommendationsAfterPipeline waits for the AI daily pipeline to finish, +// ScheduleRefreshCachedAIRecommendationsAfterPipeline waits for the AI pipeline auto run to finish, // then re-fetches ranked tenders for every company that already has a recommendation cache. func (s *companyService) ScheduleRefreshCachedAIRecommendationsAfterPipeline() { if s.aiRecommendationClient == nil { @@ -45,7 +45,7 @@ func (s *companyService) ScheduleRefreshCachedAIRecommendationsAfterPipeline() { } func (s *companyService) refreshCachedAIRecommendationsAfterPipeline(ctx context.Context) error { - if err := s.waitForPipelineDailyRunCompletion(ctx); err != nil { + if err := s.waitForPipelineAutoCompletion(ctx); err != nil { return err } @@ -95,7 +95,7 @@ func (s *companyService) refreshCachedAIRecommendationsAfterPipeline(ctx context return nil } -func (s *companyService) waitForPipelineDailyRunCompletion(ctx context.Context) error { +func (s *companyService) waitForPipelineAutoCompletion(ctx context.Context) error { if s.aiPipelineStatusClient == nil { s.logger.Info("AI pipeline status client not configured; waiting before recommendation refresh", map[string]interface{}{ "delay_sec": int(recommendationPipelineWaitInitialDelay.Seconds()), @@ -134,9 +134,9 @@ func (s *companyService) waitForPipelineDailyRunCompletion(ctx context.Context) } } - report, err := s.aiPipelineStatusClient.GetPipelineLastRun(ctx) + report, err := s.aiPipelineStatusClient.GetPipelineLastAutoRun(ctx) if err != nil { - s.logger.Warn("Failed to read AI pipeline last-run status", map[string]interface{}{ + s.logger.Warn("Failed to read AI pipeline last-auto-run status", map[string]interface{}{ "attempt": attempt, "error": err.Error(), }) @@ -147,8 +147,8 @@ func (s *companyService) waitForPipelineDailyRunCompletion(ctx context.Context) continue } - if !isPipelineDailyRunInProgress(report) { - s.logger.Info("AI pipeline daily-run finished; refreshing recommendation caches", map[string]interface{}{ + if !isPipelineRunInProgress(report) { + s.logger.Info("AI pipeline auto run finished; refreshing recommendation caches", map[string]interface{}{ "attempt": attempt, "status": strings.TrimSpace(report.Status), }) @@ -167,7 +167,7 @@ func (s *companyService) waitForPipelineDailyRunCompletion(ctx context.Context) return nil } -func isPipelineDailyRunInProgress(report *ai_summarizer.PipelineReportResponse) bool { +func isPipelineRunInProgress(report *ai_summarizer.PipelineReportResponse) bool { if report == nil { return true } diff --git a/internal/company/recommendation_pipeline_refresh_test.go b/internal/company/recommendation_pipeline_refresh_test.go index d5d7ba5..bcba8aa 100644 --- a/internal/company/recommendation_pipeline_refresh_test.go +++ b/internal/company/recommendation_pipeline_refresh_test.go @@ -6,7 +6,7 @@ import ( "tm/pkg/ai_summarizer" ) -func TestIsPipelineDailyRunInProgress(t *testing.T) { +func TestIsPipelineRunInProgress(t *testing.T) { tests := []struct { name string report *ai_summarizer.PipelineReportResponse @@ -22,8 +22,8 @@ func TestIsPipelineDailyRunInProgress(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - if got := isPipelineDailyRunInProgress(tt.report); got != tt.want { - t.Fatalf("isPipelineDailyRunInProgress() = %v, want %v", got, tt.want) + if got := isPipelineRunInProgress(tt.report); got != tt.want { + t.Fatalf("isPipelineRunInProgress() = %v, want %v", got, tt.want) } }) }