package test import ( "RenewCMS/internal/domain/purge" "RenewCMS/internal/infrastructure/scheduler" "RenewCMS/internal/infrastructure/useCases" "bytes" "context" "errors" "log" "net/http" "strings" "sync" "testing" "time" ) const ( schedulerClock = "2026-10-05T09:00:00Z" schedulerWaitTimeout = 2 * time.Second schedulerQuietPeriod = 50 * time.Millisecond ) type schedulerEnv struct { *mediaEnv runPurge *useCases.RunPurgeUseCase scheduler *useCases.PurgeSchedulerUseCase update *useCases.UpdatePurgeSettingsUseCase } func newSchedulerEnv(t *testing.T) *schedulerEnv { t.Helper() env := newPurgeEnv(t) merge := useCases.NewMergeMediaUseCase(env.images, env.articles, env.storage, newMockVariantFileRepo(), discardRecorder, plainSites(), plainAuthors()) now := func() time.Time { return env.clock } runPurge := useCases.NewRunPurgeUseCase(env.images, env.articles, merge, env.purgeSettings, now, log.New(env.logs, "", 0), discardRecorder, plainSites(), plainAuthors()) return &schedulerEnv{ mediaEnv: env, runPurge: runPurge, scheduler: useCases.NewPurgeSchedulerUseCase(env.purgeSettings, runPurge, discardRecorder), update: useCases.NewUpdatePurgeSettingsUseCase(env.purgeSettings, now), } } func (e *schedulerEnv) configure(enabled bool, interval int, enabledAt string, lastPurgeAt string) { settings := purge.Settings{AutoPurgeEnabled: enabled, AutoPurgeIntervalDays: interval} if enabledAt != "" { settings.EnabledAt = purgeTime(enabledAt) } if lastPurgeAt != "" { settings.LastPurgeAt = purgeTime(lastPurgeAt) } e.purgeSettings.settings = settings } func (e *schedulerEnv) configureDue() { e.configure(true, 10, "2026-09-20T09:00:00Z", "") } func (e *schedulerEnv) seedPlannedGroup() { e.seedGroup(hashA, 1, 2) e.useImages(1, 1, 2) } func (e *schedulerEnv) tick(t *testing.T, at string) (bool, error) { t.Helper() _, ran, err := e.scheduler.Tick(*purgeTime(at)) return ran, err } func (e *schedulerEnv) expectTick(t *testing.T, at string, wantRan bool) { t.Helper() ran, err := e.tick(t, at) if err != nil || ran != wantRan { t.Fatalf("expected ran %v and no error at %s, got %v and %v", wantRan, at, ran, err) } } func (e *schedulerEnv) expectLastPurge(t *testing.T, at string, merged int, updated int) { t.Helper() settings := e.purgeSettings.settings if settings.LastPurgeAt == nil || !settings.LastPurgeAt.Equal(*purgeTime(at)) || settings.LastPurgeMerged != merged || settings.LastPurgeArticlesUpdated != updated { t.Fatalf("unexpected last purge %+v", settings) } } func TestPurgeSchedulerDecidesWhenARunIsDue(t *testing.T) { t.Run("Automatic purge off never runs, whatever the dates are", func(t *testing.T) { env := newSchedulerEnv(t) env.configure(false, 1, "2020-01-01T00:00:00Z", "2020-01-02T00:00:00Z") env.seedPlannedGroup() env.expectTick(t, schedulerClock, false) if env.purgeSettings.saveCalls != 0 || len(env.images.deleteCalls) != 0 { t.Errorf("expected nothing to run or be saved") } }) t.Run("The defaults never run", func(t *testing.T) { env := newSchedulerEnv(t) env.seedPlannedGroup() env.expectTick(t, schedulerClock, false) if len(env.images.deleteCalls) != 0 { t.Errorf("expected no run") } }) t.Run("On, never ran, not due yet", func(t *testing.T) { env := newSchedulerEnv(t) env.configure(true, 30, "2026-09-20T09:00:00Z", "") env.seedPlannedGroup() env.expectTick(t, schedulerClock, false) if env.purgeSettings.saveCalls != 0 { t.Errorf("expected no run") } }) t.Run("On, never ran, due since enabled_at plus the interval", func(t *testing.T) { env := newSchedulerEnv(t) env.configure(true, 10, "2026-09-25T09:00:00Z", "") env.seedPlannedGroup() env.expectTick(t, schedulerClock, true) if env.purgeSettings.saveCalls != 1 || len(env.images.deleteCalls) != 1 { t.Errorf("expected exactly one run") } }) t.Run("The anchor is the later of last_purge_at and enabled_at", func(t *testing.T) { cases := []struct { name string enabledAt string lastPurgeAt string ran bool }{ {"last purge plus interval is past", "2026-09-01T00:00:00Z", "2026-09-20T00:00:00Z", true}, {"last purge plus interval is ahead", "2026-09-01T00:00:00Z", "2026-09-28T00:00:00Z", false}, {"enabled_at is later and ahead", "2026-09-28T00:00:00Z", "2026-09-01T00:00:00Z", false}, {"enabled_at is later and past", "2026-09-20T00:00:00Z", "2026-09-01T00:00:00Z", true}, {"exactly now is due", "2026-09-25T09:00:00Z", "", true}, {"one second before due", "2026-09-25T09:00:01Z", "", false}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { env := newSchedulerEnv(t) env.configure(true, 10, c.enabledAt, c.lastPurgeAt) env.expectTick(t, schedulerClock, c.ran) wantSaves := 0 if c.ran { wantSaves = 1 } if env.purgeSettings.saveCalls != wantSaves { t.Errorf("expected %d runs, got %d", wantSaves, env.purgeSettings.saveCalls) } }) } }) t.Run("Due exactly at the boundary, not before", func(t *testing.T) { env := newSchedulerEnv(t) env.configure(true, 10, "2026-09-25T09:00:00Z", "") env.expectTick(t, "2026-10-05T08:59:59Z", false) env.expectTick(t, "2026-10-05T09:00:00Z", true) }) t.Run("A run advances the anchor, so the next tick is not due", func(t *testing.T) { env := newSchedulerEnv(t) env.configureDue() env.seedPlannedGroup() env.expectTick(t, schedulerClock, true) env.expectTick(t, schedulerClock, false) if env.purgeSettings.saveCalls != 1 { t.Errorf("expected one run") } env.expectLastPurge(t, schedulerClock, 1, 1) }) t.Run("A due run that finds nothing to purge still moves the clock forward", func(t *testing.T) { env := newSchedulerEnv(t) env.configureDue() env.expectTick(t, schedulerClock, true) env.expectLastPurge(t, schedulerClock, 0, 0) env.expectTick(t, schedulerClock, false) env.expectTick(t, "2026-10-15T09:00:00Z", true) }) t.Run("A run keeps auto_purge_enabled, the interval and enabled_at", func(t *testing.T) { env := newSchedulerEnv(t) env.configureDue() before := env.purgeSettings.settings env.expectTick(t, schedulerClock, true) after := env.purgeSettings.settings if !after.AutoPurgeEnabled || after.AutoPurgeIntervalDays != before.AutoPurgeIntervalDays || !after.EnabledAt.Equal(*before.EnabledAt) { t.Errorf("unexpected settings %+v", after) } }) t.Run("Enabled with no enabled_at is not due", func(t *testing.T) { env := newSchedulerEnv(t) env.configure(true, 1, "", "") env.expectTick(t, schedulerClock, false) if env.purgeSettings.saveCalls != 0 || env.purgeSettings.settings.EnabledAt != nil { t.Errorf("expected nothing stamped or saved") } }) } func TestPurgeSchedulerFollowsSettingsChanges(t *testing.T) { updateInterval := func(env *schedulerEnv, days int) { if _, err := env.update.Update(useCases.UpdatePurgeSettingsCommand{AutoPurgeIntervalDays: &days}); err != nil { t.Fatalf("unexpected error %v", err) } } t.Run("Shortening the interval makes a run due on the next tick", func(t *testing.T) { env := newSchedulerEnv(t) env.configure(true, 30, "2026-09-25T09:00:00Z", "") env.expectTick(t, schedulerClock, false) updateInterval(env, 7) env.expectTick(t, schedulerClock, true) }) t.Run("Lengthening the interval postpones the run", func(t *testing.T) { env := newSchedulerEnv(t) env.configure(true, 10, "2026-09-20T09:00:00Z", "") updateInterval(env, 30) env.expectTick(t, schedulerClock, false) }) t.Run("Turning it off stops runs", func(t *testing.T) { env := newSchedulerEnv(t) env.configureDue() disabled := false if _, err := env.update.Update(useCases.UpdatePurgeSettingsCommand{AutoPurgeEnabled: &disabled}); err != nil { t.Fatalf("unexpected error %v", err) } env.expectTick(t, schedulerClock, false) }) t.Run("Turning it on resets the anchor to the moment of the switch", func(t *testing.T) { env := newSchedulerEnv(t) env.configure(false, 30, "2026-08-01T00:00:00Z", "2026-09-01T00:00:00Z") enabled := true if _, err := env.update.Update(useCases.UpdatePurgeSettingsCommand{AutoPurgeEnabled: &enabled}); err != nil { t.Fatalf("unexpected error %v", err) } env.expectTick(t, schedulerClock, false) env.expectTick(t, "2026-11-04T09:00:00Z", true) }) t.Run("The scheduler reads the settings at every tick", func(t *testing.T) { env := newSchedulerEnv(t) env.expectTick(t, schedulerClock, false) env.configureDue() env.expectTick(t, schedulerClock, true) env.configure(false, 10, "2026-09-20T09:00:00Z", "") env.expectTick(t, schedulerClock, false) if env.purgeSettings.getCalls < 3 { t.Errorf("expected at least three reads, got %d", env.purgeSettings.getCalls) } }) } func TestPurgeSchedulerReportsTheRunResult(t *testing.T) { t.Run("A due Tick returns what the run did", func(t *testing.T) { env := newSchedulerEnv(t) env.configureDue() env.seedPlannedGroup() result, ran, err := env.scheduler.Tick(*purgeTime(schedulerClock)) want := purge.RunResult{GroupsTotal: 1, GroupsPurged: 1, Merged: 1, ArticlesUpdated: 1} if err != nil || !ran || result != want { t.Errorf("expected %+v, got %+v, %v, %v", want, result, ran, err) } }) t.Run("A Tick that is not due returns an empty result", func(t *testing.T) { env := newSchedulerEnv(t) result, ran, err := env.scheduler.Tick(*purgeTime(schedulerClock)) if err != nil || ran || result != (purge.RunResult{}) { t.Errorf("unexpected %+v, %v, %v", result, ran, err) } }) } func TestPurgeRunKeepsSettingsChangedWhileItRuns(t *testing.T) { t.Run("A settings change made during a run survives, only last_purge_* are written by the run", func(t *testing.T) { env := newSchedulerEnv(t) env.configureDue() env.seedPlannedGroup() writing := make(chan struct{}) release := make(chan struct{}) env.purgeSettings.beforeRunWrite = func() { close(writing) <-release } runDone := make(chan error, 1) go func() { _, err := env.runPurge.Run(testCtx, nil) runDone <- err }() waitFor(t, writing) disabled := false days := 5 if _, err := env.update.Update(useCases.UpdatePurgeSettingsCommand{AutoPurgeEnabled: &disabled, AutoPurgeIntervalDays: &days}); err != nil { t.Fatalf("unexpected error %v", err) } close(release) if err := waitFor(t, runDone); err != nil { t.Fatalf("unexpected run error %v", err) } settings := env.purgeSettings.settings if settings.AutoPurgeEnabled || settings.AutoPurgeIntervalDays != 5 { t.Errorf("expected the change to survive, got %+v", settings) } env.expectLastPurge(t, schedulerClock, 1, 1) }) } func TestPurgeSchedulerFailures(t *testing.T) { t.Run("Reading the settings fails", func(t *testing.T) { env := newSchedulerEnv(t) env.configureDue() env.purgeSettings.fail.failNext("Get") ran, err := env.tick(t, schedulerClock) if ran || !errors.Is(err, errRepositoryFailure) { t.Fatalf("expected the repository error, got %v and %v", ran, err) } if len(env.images.deleteCalls) != 0 { t.Errorf("expected no run") } env.expectTick(t, schedulerClock, true) }) t.Run("RunPurge returns an error", func(t *testing.T) { env := newSchedulerEnv(t) env.configureDue() env.images.fail.failNext("ListDuplicates") ran, err := env.tick(t, schedulerClock) if ran || !errors.Is(err, errRepositoryFailure) { t.Fatalf("expected the repository error, got %v and %v", ran, err) } if env.purgeSettings.settings.LastPurgeAt != nil || env.purgeSettings.saveCalls != 0 { t.Errorf("expected last_purge_at unchanged") } env.expectTick(t, schedulerClock, true) }) t.Run("Saving the run result fails after the merges succeeded", func(t *testing.T) { env := newSchedulerEnv(t) env.configureDue() env.seedPlannedGroup() env.purgeSettings.fail.failNext("SaveLastRun") ran, err := env.tick(t, schedulerClock) if ran || !errors.Is(err, errRepositoryFailure) { t.Fatalf("expected the repository error, got %v and %v", ran, err) } if len(env.images.deleteCalls) != 1 { t.Errorf("expected the merge to stay done") } env.expectTick(t, schedulerClock, true) env.expectLastPurge(t, schedulerClock, 0, 0) }) t.Run("A run that partially fails still records what it merged", func(t *testing.T) { env := newSchedulerEnv(t) env.configureDue() env.seedGroup(hashA, 1, 2) env.seedGroup(hashB, 3, 4) env.useImages(20, 1, 2) env.useImages(21, 3, 4) env.articles.failUpdateBodyFor = map[uint32]error{21: errRepositoryFailure} env.expectTick(t, schedulerClock, true) env.expectLastPurge(t, schedulerClock, 1, 1) env.articles.failUpdateBodyFor = nil if plan := env.preview(t); numberOf(plan["groups"]) != 1 { t.Errorf("expected group B to be planned again, got %v", plan) } }) t.Run("A manual run is in progress", func(t *testing.T) { env := newSchedulerEnv(t) env.configureDue() env.seedPlannedGroup() entered := make(chan struct{}) release := make(chan struct{}) env.images.onDelete = func(uint32) { close(entered) <-release } manualDone := make(chan error, 1) go func() { _, err := env.runPurge.Run(testCtx, nil) manualDone <- err }() waitFor(t, entered) ran, err := env.tick(t, schedulerClock) if ran || !errors.Is(err, purge.ErrPurgeBusy) { t.Fatalf("expected ErrPurgeBusy, got %v and %v", ran, err) } close(release) if err := <-manualDone; err != nil { t.Fatalf("unexpected manual error %v", err) } env.expectTick(t, schedulerClock, false) }) } type schedulerJobFake struct { mu sync.Mutex calls []time.Time results []schedulerJobResult entered chan time.Time gate chan struct{} } type schedulerJobResult struct { result purge.RunResult ran bool err error } func newSchedulerJobFake(results ...schedulerJobResult) *schedulerJobFake { return &schedulerJobFake{results: results, entered: make(chan time.Time, 16)} } func (f *schedulerJobFake) Tick(now time.Time) (purge.RunResult, bool, error) { f.mu.Lock() f.calls = append(f.calls, now) var result schedulerJobResult if len(f.results) > 0 { result = f.results[0] f.results = f.results[1:] } gate := f.gate f.mu.Unlock() f.entered <- now if gate != nil { <-gate } return result.result, result.ran, result.err } func (f *schedulerJobFake) callCount() int { f.mu.Lock() defer f.mu.Unlock() return len(f.calls) } type schedulerClockFake struct { mu sync.Mutex now time.Time } func (c *schedulerClockFake) set(value string) { c.mu.Lock() defer c.mu.Unlock() c.now = *purgeTime(value) } func (c *schedulerClockFake) read() time.Time { c.mu.Lock() defer c.mu.Unlock() return c.now } type runnerHarness struct { job *schedulerJobFake clock *schedulerClockFake ticks chan time.Time startup chan time.Time logs *bytes.Buffer cancel context.CancelFunc done chan struct{} } func startRunner(t *testing.T, job *schedulerJobFake) *runnerHarness { t.Helper() h := &runnerHarness{ job: job, clock: &schedulerClockFake{}, ticks: make(chan time.Time, 8), startup: make(chan time.Time, 1), logs: &bytes.Buffer{}, done: make(chan struct{}), } h.clock.set(schedulerClock) runner := scheduler.NewRunner(job, h.clock.read, h.ticks, h.startup, log.New(h.logs, "", 0)) ctx, cancel := context.WithCancel(context.Background()) h.cancel = cancel go func() { defer close(h.done) runner.Run(ctx) }() t.Cleanup(func() { cancel() select { case <-h.done: case <-time.After(schedulerWaitTimeout): } }) return h } func (h *runnerHarness) expectStopped(t *testing.T) { t.Helper() waitFor(t, h.done) } func waitFor[T any](t *testing.T, ch <-chan T) T { t.Helper() select { case value := <-ch: return value case <-time.After(schedulerWaitTimeout): t.Fatalf("timed out waiting") panic("unreachable") } } func TestPurgeSchedulerRunner(t *testing.T) { t.Run("A failing tick does not stop the runner", func(t *testing.T) { job := newSchedulerJobFake(schedulerJobResult{err: errRepositoryFailure}, schedulerJobResult{ran: true}) h := startRunner(t, job) h.ticks <- time.Time{} waitFor(t, job.entered) time.Sleep(schedulerQuietPeriod) if job.callCount() != 1 { t.Fatalf("expected no retry between ticks, got %d calls", job.callCount()) } h.ticks <- time.Time{} waitFor(t, job.entered) h.cancel() h.expectStopped(t) if !strings.Contains(h.logs.String(), errRepositoryFailure.Error()) { t.Errorf("expected the error to be logged, got %q", h.logs.String()) } }) t.Run("A busy skip is not logged as an error", func(t *testing.T) { job := newSchedulerJobFake(schedulerJobResult{err: purge.ErrPurgeBusy}) h := startRunner(t, job) h.ticks <- time.Time{} waitFor(t, job.entered) h.cancel() h.expectStopped(t) if h.logs.Len() != 0 { t.Errorf("expected no log line, got %q", h.logs.String()) } }) t.Run("A completed run logs one line with the counts", func(t *testing.T) { cases := []struct { name string result purge.RunResult want string }{ {"a run that purged", purge.RunResult{GroupsTotal: 4, GroupsPurged: 3, Merged: 5, ArticlesUpdated: 2, FailedGroups: 1}, "scheduler: automatic purge completed: groups_purged=3 merged=5 articles_updated=2 failed_groups=1\n"}, {"an empty run", purge.RunResult{}, "scheduler: automatic purge completed: groups_purged=0 merged=0 articles_updated=0 failed_groups=0\n"}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { job := newSchedulerJobFake(schedulerJobResult{result: c.result, ran: true}) h := startRunner(t, job) h.ticks <- time.Time{} waitFor(t, job.entered) h.cancel() h.expectStopped(t) if h.logs.String() != c.want { t.Errorf("expected %q, got %q", c.want, h.logs.String()) } }) } }) t.Run("One Tick shortly after startup", func(t *testing.T) { job := newSchedulerJobFake() h := startRunner(t, job) time.Sleep(schedulerQuietPeriod) if job.callCount() != 0 { t.Fatalf("expected no Tick before the signal") } h.startup <- time.Time{} if got := waitFor(t, job.entered); !got.Equal(*purgeTime(schedulerClock)) { t.Errorf("expected the clock's time, got %s", got) } time.Sleep(schedulerQuietPeriod) if job.callCount() != 1 { t.Errorf("expected one Tick, got %d", job.callCount()) } }) t.Run("A due run is caught up at startup after downtime", func(t *testing.T) { env := newSchedulerEnv(t) env.configure(true, 30, "2026-08-01T00:00:00Z", "2026-08-26T09:00:00Z") env.seedPlannedGroup() clock := &schedulerClockFake{} clock.set(schedulerClock) ticks := make(chan time.Time, 8) startup := make(chan time.Time, 1) runner := scheduler.NewRunner(env.scheduler, clock.read, ticks, startup, log.New(env.logs, "", 0)) ctx, cancel := context.WithCancel(context.Background()) done := make(chan struct{}) go func() { defer close(done) runner.Run(ctx) }() startup <- time.Time{} deadline := time.Now().Add(schedulerWaitTimeout) for lastPurgeAt := env.purgeSettings.snapshot().LastPurgeAt; lastPurgeAt == nil || lastPurgeAt.Before(*purgeTime(schedulerClock)); lastPurgeAt = env.purgeSettings.snapshot().LastPurgeAt { if time.Now().After(deadline) { t.Fatalf("the startup run did not happen") } time.Sleep(time.Millisecond) } cancel() waitFor(t, done) if env.purgeSettings.saveCalls != 1 { t.Errorf("expected exactly one run, got %d", env.purgeSettings.saveCalls) } }) t.Run("A Tick on each hourly tick", func(t *testing.T) { job := newSchedulerJobFake() h := startRunner(t, job) for _, at := range []string{"2026-10-05T10:00:00Z", "2026-10-05T11:00:00Z", "2026-10-05T12:00:00Z"} { h.clock.set(at) h.ticks <- time.Time{} if got := waitFor(t, job.entered); !got.Equal(*purgeTime(at)) { t.Errorf("expected %s, got %s", at, got) } } if job.callCount() != 3 { t.Errorf("expected three Ticks, got %d", job.callCount()) } }) t.Run("Cancelling the context stops the runner", func(t *testing.T) { job := newSchedulerJobFake() h := startRunner(t, job) h.cancel() h.expectStopped(t) h.ticks <- time.Time{} time.Sleep(schedulerQuietPeriod) if job.callCount() != 0 { t.Errorf("expected no Tick after cancel") } }) t.Run("Cancelled before the startup signal", func(t *testing.T) { job := newSchedulerJobFake() h := startRunner(t, job) h.cancel() h.startup <- time.Time{} h.expectStopped(t) if job.callCount() != 0 { t.Errorf("expected no Tick") } }) t.Run("Shutdown waits for a run in progress without a time limit", func(t *testing.T) { env := newSchedulerEnv(t) env.configureDue() env.seedPlannedGroup() entered := make(chan struct{}) release := make(chan struct{}) env.images.onDelete = func(uint32) { close(entered) <-release } clock := &schedulerClockFake{} clock.set(schedulerClock) ticks := make(chan time.Time, 1) runner := scheduler.NewRunner(env.scheduler, clock.read, ticks, make(chan time.Time), log.New(env.logs, "", 0)) ctx, cancel := context.WithCancel(context.Background()) done := make(chan struct{}) go func() { defer close(done) runner.Run(ctx) }() ticks <- time.Time{} waitFor(t, entered) cancel() select { case <-done: t.Fatalf("Run returned while a purge was in progress") case <-time.After(schedulerQuietPeriod): } close(release) waitFor(t, done) env.expectLastPurge(t, schedulerClock, 1, 1) }) } func TestPurgeSchedulerResultIsServedBySettings(t *testing.T) { t.Run("A scheduled run updates the Last purge values served by the settings endpoint", func(t *testing.T) { env := newSchedulerEnv(t) env.configure(true, 7, "2026-09-20T09:00:00Z", "") env.seedPlannedGroup() env.expectTick(t, schedulerClock, true) rec := env.purgeSend(t, http.MethodGet, "/v1/media/purge-settings", "") body := decodeObject(t, rec.Body.String()) if body["last_purge_at"] != schedulerClock || numberOf(body["last_purge_merged"]) != 1 || numberOf(body["last_purge_articles_updated"]) != 1 || body["next_purge_at"] != "2026-10-12T09:00:00Z" || body["auto_purge_enabled"] != true || numberOf(body["auto_purge_interval_days"]) != 7 || body["enabled_at"] != "2026-09-20T09:00:00Z" { t.Errorf("unexpected body %s", rec.Body.String()) } }) t.Run("A scheduled run with nothing to purge shows zeros, not Never run", func(t *testing.T) { env := newSchedulerEnv(t) env.configureDue() env.expectTick(t, schedulerClock, true) rec := env.purgeSend(t, http.MethodGet, "/v1/media/purge-settings", "") body := decodeObject(t, rec.Body.String()) if body["last_purge_at"] != schedulerClock || numberOf(body["last_purge_merged"]) != 0 || numberOf(body["last_purge_articles_updated"]) != 0 { t.Errorf("unexpected body %s", rec.Body.String()) } }) }