package scheduler import ( "context" "database/sql" "log/slog" "sync" "time" "github.com/syncserver/internal/config" "github.com/syncserver/internal/models" "github.com/syncserver/internal/syncengine" ) type Scheduler struct { db *sql.DB engine *syncengine.Engine cfg *config.Config stopCh chan struct{} wg sync.WaitGroup } func New(database interface{ SQLDB() *sql.DB }, engine *syncengine.Engine, cfg *config.Config) *Scheduler { return &Scheduler{ db: database.SQLDB(), engine: engine, cfg: cfg, stopCh: make(chan struct{}), } } func (s *Scheduler) Start() { s.wg.Add(1) go s.run() slog.Info("scheduler started") } func (s *Scheduler) Stop() { close(s.stopCh) s.wg.Wait() slog.Info("scheduler stopped") } func (s *Scheduler) run() { defer s.wg.Done() ticker := time.NewTicker(1 * time.Minute) defer ticker.Stop() for { select { case <-s.stopCh: return case <-ticker.C: s.tick() } } } func (s *Scheduler) tick() { scheduleRepo := models.NewScheduleRepository(s.db) now := time.Now().UTC() schedules, err := scheduleRepo.GetEnabledDue(now) if err != nil { slog.Error("scheduler: failed to get due schedules", "error", err) return } for _, sch := range schedules { pairRepo := models.NewSyncPairRepository(s.db) pair, err := pairRepo.GetByID(sch.SyncPairID) if err != nil || !pair.Enabled { continue } jobID, err := s.engine.CreateJob(sch.SyncPairID, "scheduled") if err != nil { slog.Error("scheduler: failed to create job", "schedule_id", sch.ID, "error", err) continue } ctx := context.Background() go func(jobID int64, pairID int64, schID int64) { if err := s.engine.Run(ctx, jobID, pairID); err != nil { slog.Warn("scheduler: job failed", "job_id", jobID, "error", err) } expr, _ := ParseCron(sch.CronExpr) if expr != nil { next := NextRun(expr, time.Now().UTC()) scheduleRepo.UpdateNextRun(schID, next) } }(jobID, sch.SyncPairID, sch.ID) } }