thermograph/backend/daemon/internal/cron/cron.go

// Package cron is the Go replacement for notifications/scheduler.py's
// APScheduler pair (warm-cities + IndexNow). The Python side chose APScheduler
// over a Postgres-native queue because exactly one process ever runs these
// jobs — no fan-out, backpressure, or dead-letter need — and that shape
// carries over unchanged: this daemon is the sole scheduler, so plain
// per-job tickers are all the machinery required. procrastinate/pgqueuer
// remain the documented upgrade path if that ever changes — see
// docs/architecture/repo-topology-and-infrastructure.md §7.
//
// The jobs themselves stay in Python (called via the internal HTTP routes);
// this package owns only the timing.
package cron

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

// Job is one recurring task: a name for logs, an interval, and the callback
// (in practice an apiclient method hitting a /internal/jobs/* route).
type Job struct {
	Name     string
	Interval time.Duration
	Run      func(ctx context.Context) error
}

// Run drives every job on its own ticker until ctx is cancelled, then returns
// once all job goroutines have stopped. It never returns early: a failing job
// logs and keeps ticking — one bad tick (backend briefly down, timeout) must
// never kill the schedule, because nothing restarts it short of a container
// restart.
func Run(ctx context.Context, logger *slog.Logger, jobs ...Job) {
	var wg sync.WaitGroup
	for _, job := range jobs {
		wg.Add(1)
		go func(job Job) {
			defer wg.Done()
			runJob(ctx, logger, job)
		}(job)
	}
	wg.Wait()
}

func runJob(ctx context.Context, logger *slog.Logger, job Job) {
	// A plain ticker's first fire is one interval from now, NOT at startup —
	// deliberately kept from the Python scheduler (its add_job had no
	// next_run_time override for the same reason): warm-cities already runs at
	// deploy time via the deploy hook, so an immediate run on every daemon
	// boot/restart would just re-spend work (and, for warm-cities, archive-fetch
	// quota) that the deploy already paid for.
	ticker := time.NewTicker(job.Interval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			start := time.Now()
			// Run synchronously in this loop: a job can never overlap with
			// itself by construction. That matters most for warm-cities,
			// which spends the archive-fetch quota — two overlapping runs would
			// double-spend it.
			if err := job.Run(ctx); err != nil {
				if ctx.Err() != nil {
					// Shutdown raced the job; the cancellation is the story,
					// not the (expected) aborted call.
					return
				}
				logger.Error("cron job failed; will retry next tick",
					"job", job.Name, "err", err, "elapsed", time.Since(start))
			} else {
				logger.Info("cron job ok",
					"job", job.Name, "elapsed", time.Since(start))
			}
			// If the job overran its interval, a tick may have fired mid-run.
			// Skip it rather than running again back-to-back — a late tick is
			// the same double-spend as an overlapping one, just serialized.
			// (Go 1.23+ tickers already drop fires with no ready receiver;
			// this drain pins the skip semantics rather than relying on the
			// runtime's channel buffering.)
			select {
			case <-ticker.C:
				logger.Warn("cron job overran its interval; skipping the tick that fired mid-run",
					"job", job.Name, "interval", job.Interval, "elapsed", time.Since(start))
			default:
			}
		}
	}
}