package ci
import (
"fmt"
"log"
"slices"
"strings"
"time"
"craftdesign.group/quarry/internal/cronexpr"
"gopkg.in/yaml.v3"
)
// Scheduled runs are quarry's (RFC 0019's `schedule`, RFC 0021 §runs):
// the engine passes on: through uninterpreted and has no trigger that
// carries a check-in, so the forge reads each repository's default
// branch tip once a minute and launches what is due through the same
// path every other run takes. That launch registers the tip's definition
// just before it starts, so a schedule always runs the default branch's
// file, whatever a branch registered last; GitHub's rule.
//
// Cron lines are GitHub's: five fields, UTC. A minute missed while the
// forge was down is not run later, as GitHub does not.
// minScheduleGap is the most often a schedule may fire, GitHub's floor.
const minScheduleGap = 5 * time.Minute
// schedulesOf reads a ci.yml's on.schedule cron lines. A line that does
// not parse, or fires more often than every five minutes, is refused
// with its position, like every other key quarry reads.
func schedulesOf(spec []byte) ([]string, error) {
doc, ok := yamlDoc(spec)
if !ok {
return nil, nil
}
on := value(doc, "on")
if on == nil || on.Kind != yaml.MappingNode {
return nil, nil
}
list := value(on, "schedule")
if list == nil {
return nil, nil
}
if list.Kind != yaml.SequenceNode {
return nil, fmt.Errorf("line %d, column %d: on.schedule is a list of {cron: \"...\"}", list.Line, list.Column)
}
var out []string
for _, item := range list.Content {
c := value(item, "cron")
if item.Kind != yaml.MappingNode || c == nil || c.Kind != yaml.ScalarNode || len(item.Content) != 2 {
return nil, fmt.Errorf("line %d, column %d: on.schedule: each entry is {cron: \"...\"}", item.Line, item.Column)
}
if err := checkCron(c.Value); err != nil {
return nil, fmt.Errorf("line %d, column %d: on.schedule: %v", c.Line, c.Column, err)
}
out = append(out, c.Value)
}
return out, nil
}
// checkCron parses a cron line and holds it to the floor, by looking
// at a year of its fires: a gap under five minutes anywhere refuses it.
func checkCron(line string) error {
if len(strings.Fields(line)) != 5 {
return fmt.Errorf("%q is not a cron line of five fields (minute hour day month weekday, UTC)", line)
}
expr, err := cronexpr.ParseCron(line)
if err != nil {
return fmt.Errorf("%q: %v", line, err)
}
from := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
fires := expr.NextN(from, 2000)
for i := 1; i < len(fires); i++ {
if fires[i].Sub(fires[i-1]) < minScheduleGap {
return fmt.Errorf("%q fires more often than every 5 minutes, the shortest interval a schedule may have", line)
}
}
return nil
}
// due reports whether any cron line fires in minute (UTC).
func due(crons []string, minute time.Time) bool {
m := minute.UTC().Truncate(time.Minute)
for _, line := range crons {
if expr, err := cronexpr.ParseCron(line); err == nil && expr.Matches(m) {
return true
}
}
return false
}
// nextFire is the first minute after now any cron line fires.
func nextFire(crons []string, now time.Time) time.Time {
var next time.Time
for _, line := range crons {
expr, err := cronexpr.ParseCron(line)
if err != nil {
continue
}
if n := expr.NextN(now.UTC(), 1); len(n) == 1 && (next.IsZero() || n[0].Before(next)) {
next = n[0]
}
}
return next
}
// ScheduleStore is what firing needs beyond Store's.
type ScheduleStore interface {
Store
// ClaimSchedule takes one scheduled minute of one pipeline, true
// for the first claimant only: what makes a minute fire once
// across ticks, restarts and a lease handover.
ClaimSchedule(repo, pipeline string, minute time.Time) (bool, error)
// ScheduleRunning reports a scheduled run of pipeline still going.
ScheduleRunning(repo, pipeline string) bool
}
// Scheduled is one scheduled workflow as the default branch tip says.
type Scheduled struct {
Pipeline string `json:"pipeline"`
Crons []string `json:"crons"`
Next time.Time `json:"next"`
}
// Schedules lists the default branch tip's scheduled workflows, with
// each one's next fire after now.
func Schedules(r Repo, now time.Time) ([]Scheduled, error) {
c, defs, err := scheduleTip(r)
if err != nil || c.Hash == "" {
return nil, err
}
var out []Scheduled
for _, p := range defs.paths {
crons, err := scheduledIn(defs, p)
if err != nil || len(crons) == 0 {
continue
}
out = append(out, Scheduled{Pipeline: p, Crons: crons, Next: nextFire(crons, now)})
}
return out, nil
}
func scheduleTip(r Repo) (Checkin, definitions, error) {
branch := r.DefaultBranch()
tip := r.BranchTip(branch)
if tip == "" {
return Checkin{}, definitions{}, nil
}
c := Checkin{Hash: tip, Branch: branch}
defs, err := pipelinesAt(r, c)
return c, defs, err
}
// scheduledIn is a ci.yml's cron lines; a JSON pipeline has none.
func scheduledIn(defs definitions, p string) ([]string, error) {
if strings.HasSuffix(p, ".json") {
return nil, nil
}
def, err := defs.read(p)
if err != nil {
return nil, transient(fmt.Errorf("read %s: %w", p, err))
}
return schedulesOf(def)
}
// FireSchedules launches the default branch tip's workflows due in
// minute, each at most once per minute and never beside its own
// scheduled run still going. The runs have no starter: a run token is
// minted for them only where no starter letter is required.
func FireSchedules(e *Engine, st ScheduleStore, repo string, r Repo, cloneURL string, minute time.Time) error {
if !e.Enabled() {
return nil
}
minute = minute.UTC().Truncate(time.Minute)
c, defs, err := scheduleTip(r)
if err != nil || c.Hash == "" {
return err
}
var fire []string
for _, p := range defs.paths {
crons, err := scheduledIn(defs, p)
if err != nil {
return err
}
if !due(crons, minute) {
continue
}
won, err := st.ClaimSchedule(repo, p, minute)
if err != nil || !won {
if err != nil {
return err
}
continue
}
if st.ScheduleRunning(repo, p) {
log.Printf("ci: %s %s: skipped the %s schedule: its previous scheduled run is still going", repo, p, minute.Format("15:04"))
continue
}
fire = append(fire, p)
}
if len(fire) == 0 {
return nil
}
slices.Sort(fire)
return launch(e, st, repo, r, c, cloneURL, defs.only(fire), scopeSchedule)
}