package ci
import (
"fmt"
"log"
"path"
"strings"
)
// Store is the record-keeping the trigger needs; the OLTP database
// satisfies it.
type Store interface {
CIWatermark(repo string) (string, bool)
SetCIWatermark(repo, checkin string) error
AddRun(runID, repo, checkin, workflow, pipeline, status string) error
}
// Repo is the read side the trigger needs: which check-ins are new, and
// what the repository's pipelines look like AT that check-in.
type Repo interface {
// NewCheckins returns check-ins after the watermark, oldest first.
// An empty watermark means "the tip only" — a first run must not
// launch a pipeline for every check-in in a repository's history,
// which on an imported corpus would be tens of thousands.
NewCheckins(after string, limit int) ([]Checkin, error)
// Files lists a check-in's paths.
Files(version string) ([]string, error)
// Read returns one file's bytes at a check-in.
Read(path, version string) ([]byte, error)
}
// Checkin is the context a pipeline receives.
type Checkin struct {
Hash string
Branch string
Date string
User string
Comment string
}
// Trigger launches pipelines for check-ins a repository has not seen.
// It is deliberately quiet about the ordinary case — most repositories
// have no ci/ directory, and that is not a condition worth logging on
// every sync round.
func Trigger(e *Engine, st Store, repo string, r Repo, cloneURL string) error {
if !e.Enabled() {
return nil
}
mark, ok := st.CIWatermark(repo)
if !ok {
// First sight of this repository: take the tip as the watermark
// without running anything. Launching the whole history would
// be a denial of service against the engine and would say
// nothing useful about code nobody just changed.
tip, err := r.NewCheckins("", 1)
if err != nil || len(tip) == 0 {
return err
}
return st.SetCIWatermark(repo, tip[len(tip)-1].Hash)
}
fresh, err := r.NewCheckins(mark, 20)
if err != nil {
return err
}
for _, c := range fresh {
if err := launch(e, st, repo, r, c, cloneURL); err != nil {
// One bad pipeline must not wedge the watermark: the next
// check-in still gets its chance, and the failure is named.
log.Printf("ci: %s %.12s: %v", repo, c.Hash, err)
}
if err := st.SetCIWatermark(repo, c.Hash); err != nil {
return err
}
}
return nil
}
// RunAt launches a check-in's pipelines regardless of the watermark —
// the re-run path (RFC 0009 §triggers). An empty checkin means the tip.
func RunAt(e *Engine, st Store, repo string, r Repo, cloneURL, checkin string) error {
if !e.Enabled() {
return fmt.Errorf("ci: no engine configured")
}
var c Checkin
if checkin == "" {
tip, err := r.NewCheckins("", 1)
if err != nil {
return err
}
if len(tip) == 0 {
return fmt.Errorf("ci: %s has no check-ins", repo)
}
c = tip[len(tip)-1]
} else {
c = Checkin{Hash: checkin}
}
return launch(e, st, repo, r, c, cloneURL)
}
func launch(e *Engine, st Store, repo string, r Repo, c Checkin, cloneURL string) error {
files, err := r.Files(c.Hash)
if err != nil {
return err
}
var pipelines []string
for _, f := range files {
if IsPipeline(f) {
pipelines = append(pipelines, f)
}
}
if len(pipelines) == 0 {
return nil // no ci/ directory: the common case, silently fine
}
for _, p := range pipelines {
def, err := r.Read(p, c.Hash)
if err != nil {
return fmt.Errorf("read %s: %w", p, err)
}
// Registered from the definition AT THIS CHECK-IN, so a pipeline
// edit takes effect for the check-in that made it and not before.
// A ci.yml is named by its file stem — the spec language has no
// name field of its own — which is the same posture JSON
// pipelines already take on a shared engine: the register just
// before every Start keeps each run on its own definition.
var wf Workflow
if strings.HasSuffix(p, ".json") {
wf, err = e.Register(def)
} else {
wf, err = e.RegisterYAML(strings.TrimSuffix(strings.TrimSuffix(path.Base(p), ".yml"), ".yaml"), def)
}
if err != nil {
return err
}
labels := map[string]string{"repo": repo, "checkin": c.Hash, "pipeline": p}
if c.Branch != "" {
labels["branch"] = c.Branch
// A new check-in on a branch makes the branch's still-running
// builds stale; stop them (dagnats#629, best-effort).
if n, err := e.Supersede(wf.Name, map[string]string{
"repo": repo, "branch": c.Branch, "pipeline": p,
}); err == nil && n > 0 {
log.Printf("ci: %s %s: superseded %d run(s)", repo, c.Branch, n)
}
}
runID, err := e.Start(wf.Name, map[string]any{
"repo": repo,
"checkin": c.Hash,
"branch": c.Branch,
"date": c.Date,
"user": c.User,
"comment": c.Comment,
"cloneUrl": cloneURL,
"pipeline": p,
}, labels)
if err != nil {
return err
}
if err := st.AddRun(runID, repo, c.Hash, wf.Name, p, "running"); err != nil {
return err
}
log.Printf("ci: %s %.12s -> %s run %s", repo, c.Hash, wf.Name, runID)
}
return nil
}