- 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.
This commit is contained in:
Mazyar
@@ -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.
|
||||
|
||||
@@ -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"`
|
||||
}
|
||||
|
||||
|
||||
@@ -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,
|
||||
})
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user