move-data-nas/internal/scheduler/scheduler.go

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)
	}
}