diff --git a/cmd/scraper/bootstrap/bootstrap.go b/cmd/scraper/bootstrap/bootstrap.go index e21eb5a..2ad5800 100644 --- a/cmd/scraper/bootstrap/bootstrap.go +++ b/cmd/scraper/bootstrap/bootstrap.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "os" + "sync" "time" "tm/internal/notice" "tm/pkg/config" @@ -14,6 +15,9 @@ import ( "tm/ted" ) +// tedScraperRunMu ensures only one TED scraper run executes at a time (startup catch-up and cron). +var tedScraperRunMu sync.Mutex + // Init Application Configuration func InitConfig() (*Config, error) { conf, err := config.LoadConfig(".", &Config{}) @@ -101,22 +105,64 @@ func InitTEDScraper(config Config, mongoManager *mongo.ConnectionManager, appLog // start TED scraper job with error handling scheduler := schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo) + dailyJobTracker := schedule.NewDailyJobTracker(mongoManager, appLogger) + + runDailyScrape := func(trigger string) { + tedScraperRunMu.Lock() + defer tedScraperRunMu.Unlock() - function := func() { ctx := context.Background() + today := time.Now().Local() + + if trigger == "startup" { + completed, checkErr := dailyJobTracker.IsCompleted(ctx, schedule.TEDScraperJobName, today) + if checkErr != nil { + appLogger.Error("Failed to check daily scrape completion status", map[string]interface{}{ + "error": checkErr.Error(), + "date": today.Format("02/01/2006"), + }) + } else if completed { + appLogger.Info("Startup catch-up skipped: today's TED scrape already completed", map[string]interface{}{ + "date": today.Format("02/01/2006"), + }) + return + } + appLogger.Info("Running startup catch-up for today's TED scrape", map[string]interface{}{ + "date": today.Format("02/01/2006"), + }) + } else { + appLogger.Info("Running scheduled TED scrape", map[string]interface{}{ + "date": today.Format("02/01/2006"), + }) + } + err := tedScraper.Run(ctx, nil, nil) if err != nil { - appLogger.Error("Scheduled scraper run failed", map[string]interface{}{ - "error": err.Error(), + appLogger.Error("TED scraper run failed", map[string]interface{}{ + "trigger": trigger, + "date": today.Format("02/01/2006"), + "error": err.Error(), }) - // Don't exit - continue running for next scheduled execution - } else { - appLogger.Info("Scheduled scraper run completed successfully", map[string]interface{}{}) + return } + + if markErr := dailyJobTracker.MarkCompleted(ctx, schedule.TEDScraperJobName, today); markErr != nil { + appLogger.Error("Failed to mark daily scrape as completed", map[string]interface{}{ + "trigger": trigger, + "date": today.Format("02/01/2006"), + "error": markErr.Error(), + }) + } + + appLogger.Info("TED scraper run completed successfully", map[string]interface{}{ + "trigger": trigger, + "date": today.Format("02/01/2006"), + }) } + err := scheduler.AddJob(schedule.Job{ Name: "TED Scraper Job", - Func: function, + Func: func() { runDailyScrape("scheduled") }, Expr: config.TED.ScrapingInterval, }) if err != nil { @@ -126,6 +172,11 @@ func InitTEDScraper(config Config, mongoManager *mongo.ConnectionManager, appLog return nil } + // After a server restart, resume today's scrape if it was interrupted or never ran. + go func() { + runDailyScrape("startup") + }() + // Start the scheduler scheduler.Start() diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index f8c317b..94c97e8 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -158,6 +158,23 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger "interval": config.Worker.TranslationInterval, "batch_size": config.Worker.TranslationBatchSize, }) + + go func() { + appLogger.Info("Running startup catch-up for tender translations", map[string]interface{}{ + "target_languages": targetLanguages, + }) + worker := workers.NewTranslationWorker( + mongoManager, + appLogger, + tenderRepo, + aiClient, + aiStorage, + translationSuccessCounter, + targetLanguages, + config.Worker.TranslationBatchSize, + ) + worker.Run() + }() } else { appLogger.Warn("AI summarizer client not available, tender translation worker is disabled", map[string]interface{}{}) } @@ -182,8 +199,9 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger "interval": notificationInterval, }) - // Kick off one notice-processing pass without blocking startup (cron continues on schedule) + // 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{}{}) w := workers.NewNoticeWorker( mongoManager, appLogger, diff --git a/pkg/schedule/daily_job_tracker.go b/pkg/schedule/daily_job_tracker.go new file mode 100644 index 0000000..c0fc311 --- /dev/null +++ b/pkg/schedule/daily_job_tracker.go @@ -0,0 +1,105 @@ +package schedule + +import ( + "context" + "errors" + "fmt" + "time" + + "tm/pkg/logger" + "tm/pkg/mongo" + + "go.mongodb.org/mongo-driver/v2/bson" + mongodriver "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" +) + +const ( + dailyJobRunsCollection = "daily_job_runs" + DailyJobStatusCompleted = "completed" + + TEDScraperJobName = "ted_scraper" +) + +type dailyJobRun struct { + ID string `bson:"_id"` + JobName string `bson:"job_name"` + Date string `bson:"date"` + CompletedAt int64 `bson:"completed_at"` + Status string `bson:"status"` +} + +// DailyJobTracker records successful daily job runs so startup catch-up can skip +// dates that were already completed before a server restart. +type DailyJobTracker struct { + mongo *mongo.ConnectionManager + logger logger.Logger +} + +func NewDailyJobTracker(mongoManager *mongo.ConnectionManager, log logger.Logger) *DailyJobTracker { + return &DailyJobTracker{ + mongo: mongoManager, + logger: log, + } +} + +func dailyJobRunID(jobName string, date time.Time) string { + return fmt.Sprintf("%s:%s", jobName, date.Format("2006-01-02")) +} + +// IsCompleted reports whether jobName finished successfully for the given local calendar date. +func (t *DailyJobTracker) IsCompleted(ctx context.Context, jobName string, date time.Time) (bool, error) { + if jobName == "" { + return false, fmt.Errorf("daily job tracker: job name is required") + } + + var doc dailyJobRun + err := t.mongo.GetCollection(dailyJobRunsCollection).FindOne( + ctx, + bson.M{"_id": dailyJobRunID(jobName, date), "status": DailyJobStatusCompleted}, + ).Decode(&doc) + if err != nil { + if errors.Is(err, mongodriver.ErrNoDocuments) { + return false, nil + } + return false, fmt.Errorf("daily job tracker: check completed %q: %w", jobName, err) + } + + return true, nil +} + +// MarkCompleted records a successful daily job run for the given local calendar date. +func (t *DailyJobTracker) MarkCompleted(ctx context.Context, jobName string, date time.Time) error { + if jobName == "" { + return fmt.Errorf("daily job tracker: job name is required") + } + + dateKey := date.Format("2006-01-02") + now := time.Now().Unix() + + _, err := t.mongo.GetCollection(dailyJobRunsCollection).UpdateOne( + ctx, + bson.M{"_id": dailyJobRunID(jobName, date)}, + bson.M{ + "$set": bson.M{ + "job_name": jobName, + "date": dateKey, + "completed_at": now, + "status": DailyJobStatusCompleted, + }, + "$setOnInsert": bson.M{"_id": dailyJobRunID(jobName, date)}, + }, + options.UpdateOne().SetUpsert(true), + ) + if err != nil { + return fmt.Errorf("daily job tracker: mark completed %q on %s: %w", jobName, dateKey, err) + } + + t.logger.Debug("Daily job marked completed", map[string]interface{}{ + "job_name": jobName, + "date": dateKey, + "completed_at": now, + }) + + return nil +} diff --git a/pkg/schedule/daily_job_tracker_test.go b/pkg/schedule/daily_job_tracker_test.go new file mode 100644 index 0000000..f38824e --- /dev/null +++ b/pkg/schedule/daily_job_tracker_test.go @@ -0,0 +1,15 @@ +package schedule + +import ( + "testing" + "time" +) + +func TestDailyJobRunID(t *testing.T) { + date := time.Date(2026, 6, 27, 15, 30, 0, 0, time.Local) + got := dailyJobRunID(TEDScraperJobName, date) + want := "ted_scraper:2026-06-27" + if got != want { + t.Fatalf("dailyJobRunID() = %q, want %q", got, want) + } +} diff --git a/pkg/xmlparser/run_tests.sh b/pkg/xmlparser/run_tests.sh old mode 100755 new mode 100644 diff --git a/upload.sh b/upload.sh old mode 100755 new mode 100644