package scheduler import ( "RenewCMS/internal/domain/audit" "RenewCMS/internal/domain/purge" "context" "errors" "fmt" "log" "time" ) const ( TickInterval = time.Hour StartupDelay = time.Minute ) type Job interface { Tick(now time.Time) (purge.RunResult, bool, error) } type CleanupJob interface { Tick(now time.Time) (audit.CleanupResult, bool, error) } type task interface { tick(now time.Time) (summary string, ran bool, err error) } type purgeTask struct { job Job } func (p purgeTask) tick(now time.Time) (string, bool, error) { result, ran, err := p.job.Tick(now) if errors.Is(err, purge.ErrPurgeBusy) { return "", false, nil } summary := fmt.Sprintf("groups_purged=%d merged=%d articles_updated=%d failed_groups=%d", result.GroupsPurged, result.Merged, result.ArticlesUpdated, result.FailedGroups) return summary, ran, err } type cleanupTask struct { job CleanupJob } func (c cleanupTask) tick(now time.Time) (string, bool, error) { result, ran, err := c.job.Tick(now) return fmt.Sprintf("removed=%d", result.Removed), ran, err } type Runner struct { label string task task now func() time.Time ticks <-chan time.Time startup <-chan time.Time logger *log.Logger release func() } const ( purgeLabel = "automatic purge" cleanupLabel = "activity cleanup" ) func newRunner(label string, task task, now func() time.Time, ticks <-chan time.Time, startup <-chan time.Time, logger *log.Logger) *Runner { return &Runner{label: label, task: task, now: now, ticks: ticks, startup: startup, logger: logger, release: func() {}} } func NewRunner(job Job, now func() time.Time, ticks <-chan time.Time, startup <-chan time.Time, logger *log.Logger) *Runner { return newRunner(purgeLabel, purgeTask{job}, now, ticks, startup, logger) } func NewCleanupRunner(job CleanupJob, now func() time.Time, ticks <-chan time.Time, startup <-chan time.Time, logger *log.Logger) *Runner { return newRunner(cleanupLabel, cleanupTask{job}, now, ticks, startup, logger) } func NewSystemRunner(job Job, logger *log.Logger) *Runner { return withSystemTimers(func(now func() time.Time, ticks <-chan time.Time, startup <-chan time.Time) *Runner { return NewRunner(job, now, ticks, startup, logger) }) } func NewSystemCleanupRunner(job CleanupJob, logger *log.Logger) *Runner { return withSystemTimers(func(now func() time.Time, ticks <-chan time.Time, startup <-chan time.Time) *Runner { return NewCleanupRunner(job, now, ticks, startup, logger) }) } func withSystemTimers(build func(now func() time.Time, ticks <-chan time.Time, startup <-chan time.Time) *Runner) *Runner { ticker := time.NewTicker(TickInterval) startup := time.NewTimer(StartupDelay) runner := build(time.Now, ticker.C, startup.C) runner.release = func() { ticker.Stop() startup.Stop() } return runner } func (r *Runner) Run(ctx context.Context) { defer r.release() startup := r.startup for ctx.Err() == nil { select { case <-ctx.Done(): return case <-startup: startup = nil r.tick(ctx) case <-r.ticks: r.tick(ctx) } } } func (r *Runner) tick(ctx context.Context) { if ctx.Err() != nil { return } summary, ran, err := r.task.tick(r.now()) if err != nil { r.logger.Printf("scheduler: %s failed: %v", r.label, err) return } if ran { r.logger.Printf("scheduler: %s completed: %s", r.label, summary) } }