Cyrene_v2/internal/services/automations/scheduler.go

package automations

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

	"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
	"github.com/robfig/cron/v3"
)

// schedulerStopWait bounds how long Stop waits for in-flight schedule jobs to
// finish before giving up and returning. It prevents a blocked job from leaking
// the Stop caller.
const schedulerStopWait = 5 * time.Second

// ScheduleID aliases the cron entry id so callers only need this package's type.
type ScheduleID = cron.EntryID

// parserFor builds the cron spec parser matching cfg. When WithSeconds is set the
// parser accepts a leading seconds field (6-field specs); otherwise it accepts
// the standard 5-field format. Both accept descriptors such as "@every".
func parserFor(cfg Config) cron.Parser {
	if cfg.WithSeconds {
		return cron.NewParser(cron.Second | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor)
	}
	return cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor)
}

// cronLogger adapts an *slog.Logger to the cron.Logger interface so cron's own
// operational messages carry the "services/automations" source tag.
type cronLogger struct {
	log *slog.Logger
}

func (c cronLogger) Info(msg string, keysAndValues ...any) { c.log.Info(msg, keysAndValues...) }

func (c cronLogger) Error(err error, msg string, keysAndValues ...any) {
	c.log.Error(msg, append([]any{"error", err}, keysAndValues...)...)
}

// Scheduler is the cron-driven automation scheduler. It wraps a robfig/cron/v3
// scheduler and runs a registered Action each time its cron expression fires,
// publishing the outcome to the bus. It is safe for concurrent use.
type Scheduler struct {
	cfg  Config
	bus  bus.Bus
	log  *slog.Logger
	cron *cron.Cron

	mu      sync.RWMutex
	ctx     context.Context
	cancel  context.CancelFunc
	started bool
}

// New builds a Scheduler. The bus in deps is required; a nil bus returns an error
// wrapping ErrNilBus. cfg.Location defaults to time.Local.
func New(cfg Config, deps Deps) (*Scheduler, error) {
	if deps.Bus == nil {
		return nil, fmt.Errorf("automations: new scheduler: %w", ErrNilBus)
	}
	loc := cfg.Location
	if loc == nil {
		loc = time.Local
	}
	log := sourcedLog(deps.Log)
	s := &Scheduler{
		cfg: cfg,
		bus: deps.Bus,
		log: log,
		cron: cron.New(
			cron.WithLocation(loc),
			cron.WithParser(parserFor(cfg)),
			cron.WithLogger(cronLogger{log: log}),
		),
	}
	return s, nil
}

// ValidateCron reports whether expr is a valid cron expression for this
// scheduler's configured parser. A parse failure is wrapped with ErrInvalidCron.
func (s *Scheduler) ValidateCron(expr string) error {
	if _, err := parserFor(s.cfg).Parse(expr); err != nil {
		return fmt.Errorf("automations: invalid cron %q: %w: %v", expr, ErrInvalidCron, err)
	}
	return nil
}

// AddSchedule registers action to run whenever cronExpr fires, returning its
// schedule id. The cron expression is validated against the scheduler's parser;
// an invalid expression returns an error wrapping ErrInvalidCron. ctx is used
// only to bound this call; it does not govern later fires which use the
// scheduler's own live context.
func (s *Scheduler) AddSchedule(ctx context.Context, cronExpr string, action Action) (ScheduleID, error) {
	if action == nil {
		return 0, fmt.Errorf("automations: add schedule: %w", ErrNilAction)
	}
	if _, err := parserFor(s.cfg).Parse(cronExpr); err != nil {
		return 0, fmt.Errorf("automations: add schedule %q: %w: %v", cronExpr, ErrInvalidCron, err)
	}
	if ctx == nil {
		ctx = context.Background()
	}
	if err := ctx.Err(); err != nil {
		return 0, fmt.Errorf("automations: add schedule %q: %w", cronExpr, err)
	}
	id, err := s.cron.AddFunc(cronExpr, func() { s.fire(cronExpr, action) })
	if err != nil {
		return 0, fmt.Errorf("automations: add schedule %q: %w: %v", cronExpr, ErrInvalidCron, err)
	}
	s.log.Info("schedule added", "expr", cronExpr, "action", action.Name(), "id", id)
	return id, nil
}

// RemoveSchedule removes a previously added schedule by id. It is safe to call on
// an unknown id.
func (s *Scheduler) RemoveSchedule(id ScheduleID) {
	s.cron.Remove(id)
	s.log.Info("schedule removed", "id", id)
}

// Entries returns a snapshot of the registered schedules, ordered by the next run
// time. It is only populated once the scheduler has been started.
func (s *Scheduler) Entries() []cron.Entry {
	return s.cron.Entries()
}

// Entry returns the schedule registered under id, or the zero Entry when the id
// is unknown.
func (s *Scheduler) Entry(id ScheduleID) cron.Entry {
	return s.cron.Entry(id)
}

// Start begins firing registered schedules. Start derives the scheduler's live
// context from ctx; cancelling ctx (or calling Stop) stops the scheduler and
// cancels in-flight jobs. Start is idempotent.
func (s *Scheduler) Start(ctx context.Context) error {
	if ctx == nil {
		ctx = context.Background()
	}
	s.mu.Lock()
	if s.started {
		s.mu.Unlock()
		return nil
	}
	s.ctx, s.cancel = context.WithCancel(ctx)
	s.started = true
	s.mu.Unlock()
	s.cron.Start()
	s.log.Info("scheduler started")
	return nil
}

// Stop stops the scheduler, cancels in-flight jobs and waits (bounded by
// schedulerStopWait) for them to finish. It is idempotent.
func (s *Scheduler) Stop(ctx context.Context) error {
	if ctx == nil {
		ctx = context.Background()
	}
	s.mu.Lock()
	if !s.started {
		s.mu.Unlock()
		return nil
	}
	s.started = false
	if s.cancel != nil {
		s.cancel()
		s.cancel = nil
	}
	s.mu.Unlock()

	done := s.cron.Stop()
	select {
	case <-done.Done():
		s.log.Info("scheduler stopped")
		return nil
	case <-ctx.Done():
		return fmt.Errorf("automations: scheduler stop: %w", ctx.Err())
	case <-time.After(schedulerStopWait):
		s.log.Warn("scheduler stop timed out")
		return nil
	}
}

// liveCtx returns the scheduler's live context, or Background before Start.
func (s *Scheduler) liveCtx() context.Context {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if s.ctx != nil {
		return s.ctx
	}
	return context.Background()
}

// fire runs one scheduled action. It recovers panics so a misbehaving action
// cannot take down the cron scheduler goroutine, and reports its outcome to the
// bus through runAction.
func (s *Scheduler) fire(cronExpr string, action Action) {
	defer func() {
		if r := recover(); r != nil {
			s.log.Error("schedule job panicked", "expr", cronExpr, "panic", r)
		}
	}()
	runAction(s.liveCtx(), s.cfg, s.bus, s.log, "schedule", cronExpr, action, nil)
}