memby/server/internal/scheduler/scheduler.go

// Package scheduler is the gateway's reusable background-job service.
//
// Before it, every recurring job in the gateway was its own goroutine with its own
// time.Ticker, its own idea of what to do when it failed and no way for an operator to
// see whether it had run. The point of a registry is not that a ticker is hard to write —
// it is that "when did the housekeeping last run, how long did it take, and did it work"
// is a question no scattered ticker can answer, and that a job an operator cannot start
// by hand is one they can only restart the container to retry.
package scheduler

import (
	"context"
	"fmt"
	"log/slog"
	"sort"
	"sync"
	"time"

	"github.com/ponzischeme89/memby/server/internal/adminevents"
	"github.com/ponzischeme89/memby/server/internal/runtimestats"
	"github.com/ponzischeme89/memby/server/internal/store"
)

// tick is how often the scheduler looks for work. Deliberately coarse: nothing registered
// here is time-critical, and a task declaring a five-minute interval genuinely does not
// care whether it starts on the minute.
const tick = 30 * time.Second

// TaskFunc does the work. The detail it returns is the one line the console prints beside
// the run — "412 events removed" — so it should describe what happened, and be empty when
// nothing did. That emptiness is load-bearing: it is what stops a task running every ten
// minutes announcing itself into the notification bell every ten minutes.
type TaskFunc func(ctx context.Context) (detail string, err error)

// Outcome is the richer form: the sentence, plus what the run counted.
//
// It exists because "412 movies checked, 7 updated, 2 skipped" is the question an operator
// has about an integration and a sentence is a poor place to keep numbers — they cannot be
// compared between runs, sorted, or drawn as anything but prose. Most tasks count nothing
// and keep the plain TaskFunc; only work that processes a batch has anything to say here.
type Outcome struct {
	// Detail is the one line the console prints beside the run, and is empty when nothing
	// happened — the same rule TaskFunc's return follows, and for the same reason.
	Detail string
	store.RunCounts
}

// WorkFunc is a task that counts what it did. A task declares Run or Work, never both.
type WorkFunc func(ctx context.Context) (Outcome, error)

// Task is a declaration. Everything about it except the operator's overrides is code, so
// the registry is readable as a list of what the gateway does in the background.
type Task struct {
	ID          string
	Name        string
	Description string
	Group       string
	Interval    time.Duration
	// Timeout bounds one run. A housekeeping job that hangs must not hold its slot for
	// ever, and without this the only recovery is a restart.
	Timeout time.Duration
	// RunOnStart runs the task once shortly after boot regardless of when it last ran.
	// For jobs whose cost is trivial and whose value is highest immediately.
	RunOnStart bool
	// Integration names the external service this job belongs to, and is empty for the
	// gateway's own work. It is what lets the integrations area read its operational
	// history out of this registry rather than keeping a second one: a run is filed
	// against the service as well as against the job, and nothing else is needed.
	Integration string
	Run         TaskFunc
	// Work is Run's counting form. Exactly one of the two must be set.
	Work WorkFunc
}

// Status is one task as the console reads it: the declaration, the operator's overrides,
// the last run and when the next one is due.
type Status struct {
	ID          string `json:"id"`
	Name        string `json:"name"`
	Description string `json:"description"`
	Group       string `json:"group"`
	// Integration is the external service this job belongs to, empty for the gateway's
	// own work. The integrations area lists a service's jobs by matching on it.
	Integration string `json:"integration,omitempty"`
	Interval    int64  `json:"intervalSeconds"`
	// DefaultInterval is the cadence declared in code, which Interval hides whenever an
	// operator has overridden it. Both are sent because the console cannot otherwise tell
	// "every ten minutes because that is the default" from "every ten minutes because
	// somebody chose it" — and without that distinction its cadence control has no way to
	// offer a way back, or to say that a task is no longer running as shipped.
	DefaultInterval int64          `json:"defaultIntervalSeconds"`
	Enabled         bool           `json:"enabled"`
	Running         bool           `json:"running"`
	NextRun         *time.Time     `json:"nextRun,omitempty"`
	LastRun         *store.TaskRun `json:"lastRun,omitempty"`
}

type registered struct {
	task Task

	mu       sync.Mutex
	enabled  bool
	interval time.Duration
	running  bool
	nextRun  time.Time
	lastRun  *store.TaskRun
}

// effectiveInterval is the operator's override where they set one, and the declared
// interval otherwise. Zero means "never on a schedule" — a task that only ever runs when
// somebody presses the button, which is a legitimate declaration rather than a mistake.
func (r *registered) effectiveInterval() time.Duration {
	if r.interval > 0 {
		return r.interval
	}
	return r.task.Interval
}

type Scheduler struct {
	store  *store.Store
	log    *slog.Logger
	events *adminevents.Bus
	paused func() bool

	mu    sync.RWMutex
	tasks map[string]*registered
	order []string

	started bool
}

// SetPaused installs the server-wide activity gate. The function is intentionally read at
// execution time so an admin change takes effect without rebuilding the task registry.
func (s *Scheduler) SetPaused(paused func() bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.paused = paused
}

func (s *Scheduler) isPaused() bool {
	s.mu.RLock()
	paused := s.paused
	s.mu.RUnlock()
	return paused != nil && paused()
}

func New(st *store.Store, log *slog.Logger, events *adminevents.Bus) *Scheduler {
	return &Scheduler{
		store: st, log: log.With("component", "scheduler"), events: events,
		tasks: map[string]*registered{},
	}
}

// Register adds a task. Registration happens during start-up wiring, before Start, and a
// duplicate id panics rather than silently replacing: two tasks sharing an id would share
// a history and an enabled switch, and the operator would have no way to tell which of
// them the console was describing.
func (s *Scheduler) Register(task Task) {
	if s == nil {
		return
	}
	if task.ID == "" || (task.Run == nil) == (task.Work == nil) {
		panic("scheduler: a task needs an id and exactly one of Run or Work")
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if _, exists := s.tasks[task.ID]; exists {
		panic(fmt.Sprintf("scheduler: duplicate task id %q", task.ID))
	}
	if task.Timeout <= 0 {
		task.Timeout = 10 * time.Minute
	}
	s.tasks[task.ID] = &registered{task: task, enabled: true}
	s.order = append(s.order, task.ID)
}

// Start restores state and begins the loop. It returns immediately; the loop stops with
// the context.
//
// The restore is what makes a restart not a re-run. Without reading the last successful
// start time, a container replaced at three in the morning would run every overnight job
// the moment it came up, and a deployment during the day would do it again.
func (s *Scheduler) Start(ctx context.Context) {
	if s == nil {
		return
	}
	s.mu.Lock()
	if s.started {
		s.mu.Unlock()
		return
	}
	s.started = true
	s.mu.Unlock()

	s.restore(ctx)
	// Named rather than launched bare, so the console can say this worker is running
	// without having to infer it from a stack. See internal/runtimestats.
	runtimestats.Go("Task scheduler", "Scheduled jobs", func() { s.loop(ctx) })
}

func (s *Scheduler) restore(ctx context.Context) {
	if s.store == nil {
		return
	}
	if abandoned, err := s.store.AbandonRunningTasks(ctx); err != nil {
		s.log.Warn("could not close interrupted task runs", "error", err)
	} else if abandoned > 0 {
		s.log.Info("closed interrupted task runs", "runs", abandoned)
	}

	settings, err := s.store.TaskSettingsAll(ctx)
	if err != nil {
		s.log.Warn("could not read task settings", "error", err)
	}
	lastSuccess, err := s.store.TaskLastSuccess(ctx)
	if err != nil {
		s.log.Warn("could not read task history", "error", err)
	}
	latest, err := s.store.LatestTaskRuns(ctx)
	if err != nil {
		s.log.Warn("could not read latest task runs", "error", err)
	}

	now := time.Now()
	s.mu.RLock()
	defer s.mu.RUnlock()
	for id, entry := range s.tasks {
		entry.mu.Lock()
		if override, ok := settings[id]; ok {
			entry.enabled = override.Enabled
			entry.interval = time.Duration(override.IntervalSeconds) * time.Second
		}
		if run, ok := latest[id]; ok {
			copied := run
			entry.lastRun = &copied
		}
		interval := entry.effectiveInterval()
		switch {
		case entry.task.RunOnStart:
			// Shortly, not immediately: boot is already the busiest the gateway gets.
			entry.nextRun = now.Add(45 * time.Second)
		case interval <= 0:
			entry.nextRun = time.Time{}
		case !lastSuccess[id].IsZero():
			entry.nextRun = lastSuccess[id].Add(interval)
		default:
			entry.nextRun = now.Add(interval)
		}
		entry.mu.Unlock()
	}
}

func (s *Scheduler) loop(ctx context.Context) {
	ticker := time.NewTicker(tick)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			s.runDue(ctx)
		}
	}
}

func (s *Scheduler) runDue(ctx context.Context) {
	if s.isPaused() {
		return
	}
	now := time.Now()
	s.mu.RLock()
	entries := make([]*registered, 0, len(s.tasks))
	for _, entry := range s.tasks {
		entries = append(entries, entry)
	}
	s.mu.RUnlock()

	for _, entry := range entries {
		entry.mu.Lock()
		due := entry.enabled && !entry.running &&
			!entry.nextRun.IsZero() && !now.Before(entry.nextRun)
		entry.mu.Unlock()
		if due {
			go s.execute(ctx, entry, store.TriggerSchedule)
		}
	}
}

// RunNow starts a task by hand. It reports whether the task started, not whether it
// succeeded: the console watches the run history for the outcome, and a button that
// blocked until an overnight job finished would appear to hang.
func (s *Scheduler) RunNow(ctx context.Context, id string) error {
	s.mu.RLock()
	entry, ok := s.tasks[id]
	s.mu.RUnlock()
	if !ok {
		return fmt.Errorf("scheduler: no task %q", id)
	}
	if s.isPaused() {
		return fmt.Errorf("scheduler: server activity is paused for quiet time")
	}
	entry.mu.Lock()
	if entry.running {
		entry.mu.Unlock()
		return fmt.Errorf("scheduler: %s is already running", id)
	}
	entry.mu.Unlock()
	// Detached: the run outlives the admin request that asked for it, which has already
	// been answered.
	go s.execute(context.WithoutCancel(ctx), entry, store.TriggerManual)
	return nil
}

func (s *Scheduler) execute(ctx context.Context, entry *registered, trigger string) {
	if s.isPaused() {
		return
	}
	entry.mu.Lock()
	if entry.running {
		entry.mu.Unlock()
		return
	}
	entry.running = true
	task := entry.task
	interval := entry.effectiveInterval()
	entry.mu.Unlock()

	// The next run is scheduled from the start of this one rather than from its end, so a
	// task that takes four minutes on a five-minute interval keeps its cadence instead of
	// drifting a run later every time.
	started := time.Now()
	defer func() {
		entry.mu.Lock()
		entry.running = false
		if interval > 0 {
			next := started.Add(interval)
			// A run that overran its own interval goes again on the next tick rather than
			// immediately: back-to-back execution is how a slow task starves everything
			// else sharing the pool.
			if next.Before(time.Now()) {
				next = time.Now().Add(tick)
			}
			entry.nextRun = next
		} else {
			entry.nextRun = time.Time{}
		}
		entry.mu.Unlock()
	}()

	var runID int64
	if s.store != nil {
		id, err := s.store.BeginTaskRun(ctx, task.ID, task.Integration, trigger)
		if err != nil {
			s.log.Warn("could not open task run", "task", task.ID, "error", err)
		} else {
			runID = id
		}
	}

	runCtx, cancel := context.WithTimeout(ctx, task.Timeout)
	outcome, err := s.safeRun(runCtx, task)
	cancel()
	elapsed := time.Since(started)
	detail := outcome.Detail

	status := store.TaskSuccess
	failure := ""
	if err != nil {
		status, failure = store.TaskFailed, err.Error()
	}
	if s.store != nil && runID != 0 {
		if closeErr := s.store.FinishTaskRun(
			context.WithoutCancel(ctx), runID, status, detail, failure, outcome.RunCounts,
		); closeErr != nil {
			s.log.Warn("could not close task run", "task", task.ID, "error", closeErr)
		}
	}

	finished := time.Now()
	entry.mu.Lock()
	entry.lastRun = &store.TaskRun{
		ID: runID, TaskID: task.ID, IntegrationID: task.Integration,
		Trigger: trigger, Status: status,
		StartedAt: started, FinishedAt: &finished,
		DurationMS: elapsed.Milliseconds(), Detail: detail, Error: failure,
		Counts: outcome.RunCounts,
	}
	entry.mu.Unlock()

	s.announce(ctx, task, trigger, status, detail, failure, elapsed)
}

// safeRun turns a panicking task into a failed run. A background job is the one place a
// panic takes the whole process down for a reason nobody is watching for, and one
// housekeeping job with a nil map must not be able to stop the gateway serving
// television.
func (s *Scheduler) safeRun(ctx context.Context, task Task) (outcome Outcome, err error) {
	defer func() {
		if recovered := recover(); recovered != nil {
			err = fmt.Errorf("task panicked: %v", recovered)
		}
	}()
	if task.Work != nil {
		return task.Work(ctx)
	}
	detail, err := task.Run(ctx)
	return Outcome{Detail: detail}, err
}

// announce writes the log line and, when it is worth an operator's attention, publishes
// an administrative event.
//
// A *failure* is always news. A success is news only when the run reports having done
// something — the AnnounceLibrarySync rule: most passes of a housekeeping job find
// nothing, and an hourly "nothing to do" in the notification bell is what trains an
// operator to stop reading it.
func (s *Scheduler) announce(
	ctx context.Context, task Task, trigger, status, detail, failure string,
	elapsed time.Duration,
) {
	log := s.log.With("task", task.ID, "trigger", trigger,
		"duration_ms", elapsed.Milliseconds())
	if status == store.TaskFailed {
		log.Error("scheduled task failed", "error", failure)
		s.events.Publish(ctx, adminevents.Event{
			Type:     adminevents.TypeTaskFailed,
			Severity: adminevents.SeverityError,
			Title:    task.Name + " failed",
			Summary:  failure,
			Actor:    trigger,
			Target:   task.ID,
			Link:     "/admin/tasks",
			Metadata: adminevents.Meta(map[string]any{
				"taskId": task.ID, "durationMs": elapsed.Milliseconds(),
			}),
		})
		return
	}
	if detail == "" {
		log.Debug("scheduled task finished")
		return
	}
	log.Info("scheduled task finished", "detail", detail)
	s.events.Publish(ctx, adminevents.Event{
		Type:     adminevents.TypeTaskCompleted,
		Severity: adminevents.SeverityInfo,
		Title:    task.Name,
		Summary:  detail,
		Actor:    trigger,
		Target:   task.ID,
		Link:     "/admin/tasks",
		Metadata: adminevents.Meta(map[string]any{
			"taskId": task.ID, "durationMs": elapsed.Milliseconds(),
		}),
	})
}

// SetEnabled turns a task on or off and persists the choice.
func (s *Scheduler) SetEnabled(ctx context.Context, id string, enabled bool) error {
	return s.override(ctx, id, func(entry *registered) {
		entry.enabled = enabled
		if enabled && entry.nextRun.IsZero() && entry.effectiveInterval() > 0 {
			entry.nextRun = time.Now().Add(entry.effectiveInterval())
		}
	})
}

// SetInterval overrides a task's cadence. Zero restores the declared interval.
func (s *Scheduler) SetInterval(ctx context.Context, id string, interval time.Duration) error {
	if interval < 0 {
		interval = 0
	}
	if interval > 0 && interval < time.Minute {
		interval = time.Minute
	}
	return s.override(ctx, id, func(entry *registered) {
		entry.interval = interval
		if next := entry.effectiveInterval(); next > 0 {
			entry.nextRun = time.Now().Add(next)
		} else {
			entry.nextRun = time.Time{}
		}
	})
}

func (s *Scheduler) override(ctx context.Context, id string, apply func(*registered)) error {
	s.mu.RLock()
	entry, ok := s.tasks[id]
	s.mu.RUnlock()
	if !ok {
		return fmt.Errorf("scheduler: no task %q", id)
	}
	entry.mu.Lock()
	apply(entry)
	settings := store.TaskSettings{
		TaskID: id, Enabled: entry.enabled,
		IntervalSeconds: int(entry.interval / time.Second),
	}
	entry.mu.Unlock()
	if s.store == nil {
		return nil
	}
	return s.store.SetTaskSettings(ctx, settings)
}

// Snapshot is every task as the console reads it, in registration order — which is the
// order the registry is written in, and therefore groups related jobs together without
// anything having to sort them.
func (s *Scheduler) Snapshot() []Status {
	if s == nil {
		return []Status{}
	}
	s.mu.RLock()
	order := append([]string(nil), s.order...)
	tasks := make(map[string]*registered, len(s.tasks))
	for id, entry := range s.tasks {
		tasks[id] = entry
	}
	s.mu.RUnlock()

	statuses := make([]Status, 0, len(order))
	for _, id := range order {
		entry := tasks[id]
		entry.mu.Lock()
		status := Status{
			ID: entry.task.ID, Name: entry.task.Name,
			Description: entry.task.Description, Group: entry.task.Group,
			Integration:     entry.task.Integration,
			Interval:        int64(entry.effectiveInterval() / time.Second),
			DefaultInterval: int64(entry.task.Interval / time.Second),
			Enabled:         entry.enabled, Running: entry.running,
		}
		if !entry.nextRun.IsZero() && entry.enabled {
			next := entry.nextRun
			status.NextRun = &next
		}
		if entry.lastRun != nil {
			last := *entry.lastRun
			status.LastRun = &last
		}
		entry.mu.Unlock()
		statuses = append(statuses, status)
	}
	return statuses
}

// Groups lists the distinct task groups, so the console can lay the registry out in
// sections without a hard-coded list that drifts from the tasks themselves.
func (s *Scheduler) Groups() []string {
	seen := map[string]bool{}
	groups := []string{}
	for _, status := range s.Snapshot() {
		if status.Group != "" && !seen[status.Group] {
			seen[status.Group] = true
			groups = append(groups, status.Group)
		}
	}
	sort.Strings(groups)
	return groups
}