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 {