From 492f9ba3c87f1e0a42c3407f118e992dbe943a78 Mon Sep 17 00:00:00 2001 From: Mazyar Date: Wed, 1 Jul 2026 19:42:36 +0330 Subject: [PATCH] Implement AI pipeline auto worker functionality - 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. --- cmd/worker/bootstrap/bootstrap.go | 86 ++++++++++++++++++++++++++ cmd/worker/bootstrap/config.go | 5 ++ cmd/worker/workers/ai_pipeline_auto.go | 56 +++++++++++++++++ pkg/schedule/daily_job_tracker.go | 3 +- 4 files changed, 149 insertions(+), 1 deletion(-) create mode 100644 cmd/worker/workers/ai_pipeline_auto.go diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index 94c97e8..f385f00 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -1,8 +1,10 @@ package bootstrap import ( + "context" "fmt" "strings" + "sync" "time" "tm/cmd/worker/workers" "tm/internal/notice" @@ -17,6 +19,9 @@ 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 + // Init Application Configuration func InitConfig() (*Config, error) { conf, err := config.LoadConfig(".", &Config{}) @@ -89,6 +94,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, }) // Initialize repositories noticeRepo := notice.NewRepository(mongoManager, appLogger) @@ -199,6 +206,85 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger "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{}{}) diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index 4a7bb6f..0db645f 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -35,6 +35,11 @@ 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. + // 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 string `env:"WORKER_AI_PIPELINE_AUTO_INTERVAL" envDefault:"0 0 11 * * *"` } // AISummarizerConfig holds configuration for the external AI summarizer service. diff --git a/cmd/worker/workers/ai_pipeline_auto.go b/cmd/worker/workers/ai_pipeline_auto.go new file mode 100644 index 0000000..a24407c --- /dev/null +++ b/cmd/worker/workers/ai_pipeline_auto.go @@ -0,0 +1,56 @@ +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/pkg/schedule/daily_job_tracker.go b/pkg/schedule/daily_job_tracker.go index c0fc311..1f03a74 100644 --- a/pkg/schedule/daily_job_tracker.go +++ b/pkg/schedule/daily_job_tracker.go @@ -18,7 +18,8 @@ const ( dailyJobRunsCollection = "daily_job_runs" DailyJobStatusCompleted = "completed" - TEDScraperJobName = "ted_scraper" + TEDScraperJobName = "ted_scraper" + AIPipelineAutoJobName = "ai_pipeline_auto" ) type dailyJobRun struct {