From 51a1a6aa828058fd433057c424862b8127c5379d Mon Sep 17 00:00:00 2001 From: Mazyar Date: Mon, 13 Jul 2026 11:54:58 +0330 Subject: [PATCH] Refactor AI pipeline terminology for consistency - Updated comments and logging messages in the worker and related files to replace "daily-run" with "auto run" for clarity and consistency. - Adjusted the `WorkerConfig` struct to reflect the new terminology in configuration settings. - Renamed functions and test cases to align with the updated terminology, enhancing code readability and maintainability. This change improves the clarity of the AI pipeline's functionality within the tender management system. --- cmd/worker/bootstrap/bootstrap.go | 32 +++++++++---------- cmd/worker/bootstrap/config.go | 6 ++-- cmd/worker/workers/ai_pipeline_daily.go | 16 +++++----- .../recommendation_pipeline_refresh.go | 18 +++++------ .../recommendation_pipeline_refresh_test.go | 6 ++-- 5 files changed, 39 insertions(+), 39 deletions(-) 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) } }) }