131 lines
3.3 KiB
Go
131 lines
3.3 KiB
Go
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)
|
|
}
|
|
}
|