From 1ad0206e61e8a8d548af70ae94a08ecddd2a0142 Mon Sep 17 00:00:00 2001 From: Mazyar Date: Tue, 7 Jul 2026 12:17:03 +0330 Subject: [PATCH] Refactor AI pipeline handling to support daily runs - 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. --- cmd/worker/bootstrap/bootstrap.go | 64 ++++++++++++++----------- cmd/worker/bootstrap/config.go | 4 +- cmd/worker/workers/ai_pipeline_auto.go | 56 ---------------------- cmd/worker/workers/ai_pipeline_daily.go | 52 ++++++++++++++++++++ pkg/schedule/daily_job_tracker.go | 4 +- 5 files changed, 92 insertions(+), 88 deletions(-) delete mode 100644 cmd/worker/workers/ai_pipeline_auto.go create mode 100644 cmd/worker/workers/ai_pipeline_daily.go diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index f385f00..ed6081a 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -2,6 +2,7 @@ package bootstrap import ( "context" + "errors" "fmt" "strings" "sync" @@ -19,8 +20,8 @@ import ( "tm/pkg/schedule" ) -// aiPipelineAutoRunMu ensures only one AI pipeline auto run executes at a time (startup catch-up and cron). -var aiPipelineAutoRunMu sync.Mutex +// 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) { @@ -94,8 +95,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_auto_enabled": config.Worker.AIPipelineAutoEnabled, - "ai_pipeline_auto_interval": config.Worker.AIPipelineAutoInterval, + "ai_pipeline_daily_enabled": config.Worker.AIPipelineAutoEnabled, + "ai_pipeline_daily_interval": config.Worker.AIPipelineAutoInterval, }) // Initialize repositories noticeRepo := notice.NewRepository(mongoManager, appLogger) @@ -207,28 +208,28 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger }) if !config.Worker.AIPipelineAutoEnabled { - appLogger.Info("AI pipeline auto worker disabled by configuration", map[string]interface{}{ - "ai_pipeline_auto_enabled": false, + 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) - runAIPipelineAuto := func(trigger string) { - aiPipelineAutoRunMu.Lock() - defer aiPipelineAutoRunMu.Unlock() + runAIPipelineDaily := func(trigger string) { + aiPipelineDailyRunMu.Lock() + defer aiPipelineDailyRunMu.Unlock() ctx := context.Background() today := time.Now().Local() - completed, checkErr := dailyJobTracker.IsCompleted(ctx, schedule.AIPipelineAutoJobName, today) + completed, checkErr := dailyJobTracker.IsCompleted(ctx, schedule.AIPipelineDailyJobName, today) if checkErr != nil { - appLogger.Error("Failed to check AI pipeline auto completion status", map[string]interface{}{ + 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 auto skipped: already completed today", map[string]interface{}{ + appLogger.Info("AI pipeline daily-run skipped: already completed today", map[string]interface{}{ "trigger": trigger, "date": today.Format("02/01/2006"), }) @@ -236,53 +237,60 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger } if trigger == "startup" { - appLogger.Info("Running startup catch-up for today's AI pipeline auto", map[string]interface{}{ + 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 auto", map[string]interface{}{ + appLogger.Info("Running scheduled AI pipeline daily-run", map[string]interface{}{ "date": today.Format("02/01/2006"), }) } - worker := workers.NewAIPipelineAutoWorker(appLogger, aiClient) + worker := workers.NewAIPipelineDailyWorker(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(), - }) + 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.AIPipelineAutoJobName, today); markErr != nil { - appLogger.Error("Failed to mark AI pipeline auto as completed", map[string]interface{}{ + 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 auto run completed successfully", map[string]interface{}{ + 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 Auto Job", - Func: func() { runAIPipelineAuto("scheduled") }, + Name: "AI Pipeline Daily Run Job", + Func: func() { runAIPipelineDaily("scheduled") }, Expr: config.Worker.AIPipelineAutoInterval, }) - appLogger.Info("Scheduled AI pipeline auto worker", map[string]interface{}{ + appLogger.Info("Scheduled AI pipeline daily-run worker", map[string]interface{}{ "interval": config.Worker.AIPipelineAutoInterval, }) go func() { - runAIPipelineAuto("startup") + runAIPipelineDaily("startup") }() } else { - appLogger.Warn("AI summarizer client not available, AI pipeline auto worker is disabled", map[string]interface{}{}) + 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. diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index 0db645f..18606b8 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -35,10 +35,10 @@ 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 auto run on the worker. + // AIPipelineAutoEnabled schedules the daily AI pipeline daily-run on the worker. // On-demand pipeline routes on web are unaffected. AIPipelineAutoEnabled bool `env:"WORKER_AI_PIPELINE_AUTO_ENABLED" envDefault:"true"` - // AIPipelineAutoInterval runs once per day by default at 11:00 UTC (after TED scrape at 10:00). + // AIPipelineAutoInterval runs daily-run 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 * * *"` } diff --git a/cmd/worker/workers/ai_pipeline_auto.go b/cmd/worker/workers/ai_pipeline_auto.go deleted file mode 100644 index a24407c..0000000 --- a/cmd/worker/workers/ai_pipeline_auto.go +++ /dev/null @@ -1,56 +0,0 @@ -package workers - -import ( - "context" - "errors" - "time" - - "tm/pkg/ai_summarizer" - "tm/pkg/logger" -) - -// AIPipelineAutoWorker triggers the Opplens AI pipeline auto run (sync + complete missing work). -type AIPipelineAutoWorker struct { - Logger logger.Logger - AIClient *ai_summarizer.Client -} - -// NewAIPipelineAutoWorker creates a worker that calls POST /pipeline/auto on the AI service. -func NewAIPipelineAutoWorker(log logger.Logger, aiClient *ai_summarizer.Client) *AIPipelineAutoWorker { - return &AIPipelineAutoWorker{ - Logger: log, - AIClient: aiClient, - } -} - -// Run starts the pipeline auto job. A 409 response means a run is already in progress and is not an error. -func (w *AIPipelineAutoWorker) Run() error { - if w.AIClient == nil { - 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 auto worker started", map[string]interface{}{}) - - resp, err := w.AIClient.PipelineAuto(ctx) - if err != nil { - if errors.Is(err, ai_summarizer.ErrPipelineJobRunning) { - w.Logger.Info("AI pipeline auto skipped: job already running", map[string]interface{}{}) - return nil - } - w.Logger.Error("AI pipeline auto worker failed", map[string]interface{}{ - "error": err.Error(), - }) - return err - } - - w.Logger.Info("AI pipeline auto worker triggered successfully", map[string]interface{}{ - "status": resp.Status, - "report": resp.Report, - }) - - return nil -} diff --git a/cmd/worker/workers/ai_pipeline_daily.go b/cmd/worker/workers/ai_pipeline_daily.go new file mode 100644 index 0000000..d3684df --- /dev/null +++ b/cmd/worker/workers/ai_pipeline_daily.go @@ -0,0 +1,52 @@ +package workers + +import ( + "context" + "time" + + "tm/pkg/ai_summarizer" + "tm/pkg/logger" +) + +// AIPipelineDailyWorker triggers the Opplens AI pipeline daily run (sync + full pipeline on newly synced tenders). +type AIPipelineDailyWorker struct { + Logger logger.Logger + AIClient *ai_summarizer.Client +} + +// NewAIPipelineDailyWorker creates a worker that calls POST /pipeline/daily-run on the AI service. +func NewAIPipelineDailyWorker(log logger.Logger, aiClient *ai_summarizer.Client) *AIPipelineDailyWorker { + return &AIPipelineDailyWorker{ + Logger: log, + AIClient: aiClient, + } +} + +// Run starts the pipeline daily-run 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{}{}) + return nil + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + + w.Logger.Info("AI pipeline daily worker started", map[string]interface{}{}) + + resp, err := w.AIClient.PipelineDailyRun(ctx) + if err != nil { + w.Logger.Error("AI pipeline daily worker failed", map[string]interface{}{ + "error": err.Error(), + }) + return err + } + + w.Logger.Info("AI pipeline daily worker triggered successfully", map[string]interface{}{ + "status": resp.Status, + "report": resp.Report, + }) + + return nil +} diff --git a/pkg/schedule/daily_job_tracker.go b/pkg/schedule/daily_job_tracker.go index 1f03a74..1029291 100644 --- a/pkg/schedule/daily_job_tracker.go +++ b/pkg/schedule/daily_job_tracker.go @@ -18,8 +18,8 @@ const ( dailyJobRunsCollection = "daily_job_runs" DailyJobStatusCompleted = "completed" - TEDScraperJobName = "ted_scraper" - AIPipelineAutoJobName = "ai_pipeline_auto" + TEDScraperJobName = "ted_scraper" + AIPipelineDailyJobName = "ai_pipeline_daily" ) type dailyJobRun struct {