// Package jobs provides a centralized cron scheduler for all background // maintenance tasks. It wraps github.com/robfig/cron/v3 with per-job // concurrency control, timeout, panic recovery, and logging. // // Usage: // // sched := jobs.New() // jobs.RegisterAll(sched) // sched.Start() // defer sched.Shutdown() package jobs import ( "context" "crypto/rand" "crussell/clock" "crussell/internal/logutil" "encoding/base64" "fmt" "log" "os" "strings" "sync/atomic" "time" "github.com/robfig/cron/v3" ) var ( hostname string randomPrefix string jobID atomic.Uint64 ) func init() { initHostname(os.Hostname) initRandomPrefix() } // initHostname resolves the hostname, falling back to "localhost" on error. // It accepts a getHostname parameter so tests can inject failures. func initHostname(getHostname func() (string, error)) { h, err := getHostname() if err != nil || h == "" { h = "localhost" } hostname = h } func initRandomPrefix() { var buf [12]byte if _, err := rand.Read(buf[:]); err != nil { panic("crypto/rand.Read failed: " + err.Error()) } b64 := base64.StdEncoding.EncodeToString(buf[:]) b64 = strings.NewReplacer("+", "", "/", "").Replace(b64) randomPrefix = b64[0:10] } func RandomPrefix() string { return randomPrefix } func Hostname() string { return hostname } // Shorthand references to logutil constants (avoids package-qualified noise). var ( colorReset = logutil.Reset colorDim = logutil.Dim colorCyan = logutil.Cyan colorGreen = logutil.Green colorRed = logutil.Red colorYellow = logutil.Yellow colorMagenta = logutil.Magenta colorBoldMagenta = logutil.BoldMagenta ) var ( coloredDuration = logutil.ColoredDuration coloredRows = logutil.ColoredRows ) // Job defines a single periodic task. type Job struct { // Name is a human-readable name for logging. Name string // Schedule is a standard 5-field cron expression ("*/5 * * * *"). Schedule string // Timeout is the maximum duration per execution. 0 = no timeout. Timeout time.Duration // Concurrency limits simultaneous runs. 0 = unlimited, 1 = serial (skip if inflight). Concurrency int // Handler is the work function. It receives a context that is cancelled on shutdown. // Returns the number of rows affected (inserted/updated/deleted) and any error. Handler func(context.Context) (int, error) } // Scheduler manages all registered cron jobs with parallel execution. type Scheduler struct { cron *cron.Cron entries []cron.EntryID registry []Job baseCtx context.Context cancel context.CancelFunc semaphores map[string]chan struct{} } // New creates a new Scheduler. func New() *Scheduler { ctx, cancel := context.WithCancel(context.Background()) return &Scheduler{ cron: cron.New(cron.WithLocation(clock.London)), baseCtx: ctx, cancel: cancel, semaphores: make(map[string]chan struct{}), } } // Register adds a job to the scheduler. Must be called before Start. // Panics if the cron expression is invalid. func (s *Scheduler) Register(job Job) { parser := cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow) if _, err := parser.Parse(job.Schedule); err != nil { panic("jobs: invalid cron schedule \"" + job.Schedule + "\" for job \"" + job.Name + "\": " + err.Error()) } s.registry = append(s.registry, job) } // Start begins executing all registered jobs on their schedules. func (s *Scheduler) Start() { for _, job := range s.registry { j := job entryID, err := s.cron.AddFunc(j.Schedule, s.wrapJob(j)) if err != nil { log.Fatalf("jobs: failed to add func for %q: %v", j.Name, err) } s.entries = append(s.entries, entryID) } s.cron.Start() log.Printf("jobs: scheduler started — %d jobs registered", len(s.registry)) } // Shutdown gracefully stops the scheduler. It prevents new job invocations // and cancels the base context for in-flight jobs. Returns a channel that // closes when all in-flight jobs complete. func (s *Scheduler) Shutdown() <-chan struct{} { log.Println("jobs: scheduler shutting down...") s.cancel() // Cancel base context — in-flight handlers should abort ctx := s.cron.Stop() return ctx.Done() } // wrapJob adds panic recovery, timeout, concurrency control, and logging. func (s *Scheduler) wrapJob(job Job) func() { var sem chan struct{} if job.Concurrency > 0 { sem = make(chan struct{}, job.Concurrency) } return func() { jid := fmt.Sprintf("%s/%s-%06d", hostname, randomPrefix, jobID.Add(1)) jidStr := colorYellow + jid + colorReset // Concurrency guard: if max runs are in-flight, skip this tick. if sem != nil { select { case sem <- struct{}{}: // Acquired slot default: log.Printf("%s[WARN]%s %s \"%sJOB %s%s%s\" %sskipped%s (previous run still in progress)", colorYellow, colorReset, jidStr, colorBoldMagenta, colorYellow, job.Name, colorReset, colorDim, colorReset) return } defer func() { <-sem }() } // Panic recovery so one bad job can't take down the scheduler. defer func() { if r := recover(); r != nil { log.Printf("%s[ERROR]%s %s \"%sJOB %s%s%s\" %spanic recovered%s: %v", colorMagenta, colorReset, jidStr, colorBoldMagenta, colorMagenta, job.Name, colorReset, colorYellow, colorReset, r) } }() start := time.Now() log.Printf("%s[DEBUG]%s %s \"%sJOB %s%s%s\" %sstarted%s", colorCyan, colorReset, jidStr, colorBoldMagenta, colorCyan, job.Name, colorReset, colorDim, colorReset) ctx := s.baseCtx var cancel context.CancelFunc if job.Timeout > 0 { ctx, cancel = context.WithTimeout(ctx, job.Timeout) defer cancel() } n, err := job.Handler(ctx) switch { case err != nil: log.Printf("%s[ERROR]%s %s \"%sJOB %s%s%s\" %sfailed%s: %v in %s", colorRed, colorReset, jidStr, colorBoldMagenta, colorRed, job.Name, colorReset, colorDim, colorReset, err, coloredDuration(time.Since(start))) case n > 0: log.Printf("%s[INFO]%s %s \"%sJOB %s%s%s\" %scompleted%s: %d %s in %s", colorGreen, colorReset, jidStr, colorBoldMagenta, colorGreen, job.Name, colorReset, colorDim, colorReset, n, coloredRows(n), coloredDuration(time.Since(start))) default: log.Printf("%s[DEBUG]%s %s \"%sJOB %s%s%s\" %scompleted%s: %s in %s", colorCyan, colorReset, jidStr, colorBoldMagenta, colorGreen, job.Name, colorReset, colorDim, colorReset, coloredRows(0), coloredDuration(time.Since(start))) } } }