feat: media duplicate purge, part 2 of 3: the automatic purge scheduler (Go only)

internal/infrastructure/persistence/repositories/purgeSettingsRepository.go

@@ -5,6 +5,7 @@ import (
 	"RenewCMS/internal/infrastructure/persistence/mappers"
 	entity "RenewCMS/internal/infrastructure/persistence/models"
 	"errors"
+	"time"
 
 	"gorm.io/gorm"
 )
@@ -38,3 +39,19 @@ func (p *PurgeSettingsRepository) Save(settings domainPurge.Settings) error {
 	model.ID = purgeSettingsRowID
 	return p.db.Save(&model).Error
 }
+
+func (p *PurgeSettingsRepository) SaveLastRun(at time.Time, merged int, articlesUpdated int) error {
+	result := p.db.Model(&entity.PurgeSettings{}).Where("id = ?", purgeSettingsRowID).Updates(map[string]any{
+		"last_purge_at":               at,
+		"last_purge_merged":           merged,
+		"last_purge_articles_updated": articlesUpdated,
+	})
+	if result.Error != nil || result.RowsAffected > 0 {
+		return result.Error
+	}
+	settings := domainPurge.DefaultSettings()
+	settings.LastPurgeAt = &at
+	settings.LastPurgeMerged = merged
+	settings.LastPurgeArticlesUpdated = articlesUpdated
+	return p.Save(settings)
+}

internal/infrastructure/scheduler/runner.go

@@ -0,0 +1,75 @@
+package scheduler
+
+import (
+	"RenewCMS/internal/domain/purge"
+	"context"
+	"errors"
+	"log"
+	"time"
+)
+
+const (
+	TickInterval = time.Hour
+	StartupDelay = time.Minute
+)
+
+type Job interface {
+	Tick(now time.Time) (purge.RunResult, bool, error)
+}
+
+type Runner struct {
+	job     Job
+	now     func() time.Time
+	ticks   <-chan time.Time
+	startup <-chan time.Time
+	logger  *log.Logger
+	release func()
+}
+
+func NewRunner(job Job, now func() time.Time, ticks <-chan time.Time, startup <-chan time.Time, logger *log.Logger) *Runner {
+	return &Runner{job: job, now: now, ticks: ticks, startup: startup, logger: logger, release: func() {}}
+}
+
+func NewSystemRunner(job Job, logger *log.Logger) *Runner {
+	ticker := time.NewTicker(TickInterval)
+	startup := time.NewTimer(StartupDelay)
+	runner := NewRunner(job, time.Now, ticker.C, startup.C, logger)
+	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
+	}
+	result, ran, err := r.job.Tick(r.now())
+	if errors.Is(err, purge.ErrPurgeBusy) {
+		return
+	}
+	if err != nil {
+		r.logger.Printf("scheduler: automatic purge failed: %v", err)
+		return
+	}
+	if ran {
+		r.logger.Printf("scheduler: automatic purge completed: groups_purged=%d merged=%d articles_updated=%d failed_groups=%d", result.GroupsPurged, result.Merged, result.ArticlesUpdated, result.FailedGroups)
+	}
+}

internal/infrastructure/useCases/PurgeScheduler.go

@@ -0,0 +1,35 @@
+package useCases
+
+import (
+	"RenewCMS/internal/domain/purge"
+	"time"
+)
+
+type PurgeRunner interface {
+	Run(fingerprint *string) (purge.RunResult, error)
+}
+
+type PurgeSchedulerUseCase struct {
+	repository purge.Repository
+	runner     PurgeRunner
+}
+
+func NewPurgeSchedulerUseCase(repository purge.Repository, runner PurgeRunner) *PurgeSchedulerUseCase {
+	return &PurgeSchedulerUseCase{repository, runner}
+}
+
+func (u *PurgeSchedulerUseCase) Tick(now time.Time) (purge.RunResult, bool, error) {
+	settings, err := u.repository.Get()
+	if err != nil {
+		return purge.RunResult{}, false, err
+	}
+	next := settings.NextPurgeAt()
+	if next == nil || now.Before(*next) {
+		return purge.RunResult{}, false, nil
+	}
+	result, err := u.runner.Run(nil)
+	if err != nil {
+		return purge.RunResult{}, false, err
+	}
+	return result, true, nil
+}

internal/infrastructure/useCases/RunPurge.go

@@ -10,14 +10,6 @@ import (
 	"time"
 )
 
-type RunPurgeResult struct {
-	GroupsTotal     int
-	GroupsPurged    int
-	Merged          int
-	ArticlesUpdated int
-	FailedGroups    int
-}
-
 type RunPurgeUseCase struct {
 	planner    *purgePlanner
 	merge      *MergeMediaUseCase
@@ -37,30 +29,30 @@ func NewRunPurgeUseCase(imageRepository image.Repository, articleRepository arti
 	}
 }
 
-func (u *RunPurgeUseCase) Run(fingerprint *string) (RunPurgeResult, error) {
+func (u *RunPurgeUseCase) Run(fingerprint *string) (purge.RunResult, error) {
 	if !u.guard.TryLock() {
-		return RunPurgeResult{}, purge.ErrPurgeBusy
+		return purge.RunResult{}, purge.ErrPurgeBusy
 	}
 	defer u.guard.Unlock()
 
 	plan, err := u.planner.plan()
 	if err != nil {
-		return RunPurgeResult{}, err
+		return purge.RunResult{}, err
 	}
 	if fingerprint != nil && *fingerprint != plan.fingerprint {
-		return RunPurgeResult{}, purge.ErrPurgeConflict
+		return purge.RunResult{}, purge.ErrPurgeConflict
 	}
 
 	result := u.mergeGroups(plan)
 	if err := u.recordRun(result); err != nil {
 		u.logger.Printf("purge: could not record the run result: %v", err)
-		return RunPurgeResult{}, err
+		return purge.RunResult{}, err
 	}
 	return result, nil
 }
 
-func (u *RunPurgeUseCase) mergeGroups(plan purgePlan) RunPurgeResult {
-	result := RunPurgeResult{GroupsTotal: len(plan.groups)}
+func (u *RunPurgeUseCase) mergeGroups(plan purgePlan) purge.RunResult {
+	result := purge.RunResult{GroupsTotal: len(plan.groups)}
 	updated := map[uint32]bool{}
 	for _, group := range plan.groups {
 		merged, err := u.merge.Merge(group.kept.ID, slices.Clone(group.mergedIds))
@@ -78,14 +70,6 @@ func (u *RunPurgeUseCase) mergeGroups(plan purgePlan) RunPurgeResult {
 	return result
 }
 
-func (u *RunPurgeUseCase) recordRun(result RunPurgeResult) error {
-	settings, err := u.repository.Get()
-	if err != nil {
-		return err
-	}
-	at := u.now().UTC()
-	settings.LastPurgeAt = &at
-	settings.LastPurgeMerged = result.Merged
-	settings.LastPurgeArticlesUpdated = result.ArticlesUpdated
-	return u.repository.Save(settings)
+func (u *RunPurgeUseCase) recordRun(result purge.RunResult) error {
+	return u.repository.SaveLastRun(u.now().UTC(), result.Merged, result.ArticlesUpdated)
 }