Crussell/backend/internal/jobs/scheduler.go

// 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() {
	h, err := os.Hostname()
	if err != nil || h == "" {
		h = "localhost"
	}
	hostname = h

	var buf [12]byte
	rand.Read(buf[:])
	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
	colorBold        = logutil.Bold
	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)
		if 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)))
		} else if 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)))
		} else {
			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)))
		}
	}
}