// Package ci is 0001 §6's trigger and display: dagnats is the engine,
// quarry launches runs and shows them.
//
// The division matters. quarry does not define pipelines, schedule
// steps, retry, or run anything — a pipeline is TRACKED CONTENT
// (`ci/*.json` in the repository, dagnats' native workflow format), so
// the pipeline for a historical check-in is the pipeline that ran, and
// there is no forge-side pipeline store to drift from the code. Workers
// belong to the project, reached over dagnats' HTTP bridge in whatever
// language it likes.
//
// What quarry owns is narrow: notice a check-in, find its pipeline at
// THAT check-in, start the run, remember which check-in it was for, and
// show the result.
//
// This package is a data plane; the arch gate restricts its importers.
package ci
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/url"
"path"
"strconv"
"strings"
"time"
)
// Dir is where a repository keeps its pipelines. Tracked content, read
// at the check-in under test.
const Dir = "ci/"
// Engine talks to dagnats' control plane.
type Engine struct {
base string
http *http.Client
}
// New returns an engine client, or a disabled one when base is empty
// (no --dagnats configured), whose calls are no-ops.
func New(base string) *Engine {
return &Engine{base: strings.TrimRight(base, "/"),
http: &http.Client{Timeout: 10 * time.Second}}
}
// Enabled reports whether CI is configured at all.
func (e *Engine) Enabled() bool { return e != nil && e.base != "" }
// Workflow is the part of dagnats' workflow document quarry needs to
// read. The rest is passed through untouched: quarry is not a schema
// authority for someone else's format, and validating it here would
// mean rejecting pipelines dagnats accepts.
type Workflow struct {
Name string `json:"name"`
}
// Run is one dagnats run as its API reports it.
type Run struct {
RunID string `json:"run_id"`
Workflow string `json:"workflow_id"`
Labels map[string]string `json:"labels,omitempty"`
Status string `json:"status"`
Steps map[string]Step `json:"steps"`
CreatedAt string `json:"created_at"`
CompletedAt string `json:"completed_at,omitempty"`
// Input echoes what the run was started with — repo, checkin,
// cloneUrl — which is what a repro line needs (RFC 0009 §logs).
Input map[string]any `json:"input,omitempty"`
}
// Step is one step's state within a run. The timestamps are dagnats#626
// (engine-stamped dispatch and terminal times); absent on an engine
// that predates it, and the page says nothing rather than zero.
type Step struct {
Status string `json:"status"`
Attempts int `json:"attempts"`
Error string `json:"error,omitempty"`
StartedAt string `json:"started_at,omitempty"`
CompletedAt string `json:"completed_at,omitempty"`
}
// LogChunk is one bounded piece of a step's output — dagnats#624's
// hot-lane contract: {seq, ts, stream: out|err|marker, data}. A
// marker chunk records an event (failed) rather than output.
type LogChunk struct {
Seq int `json:"seq"`
TS string `json:"ts"`
Stream string `json:"stream"`
Data string `json:"data"`
// Step is quarry's addition when chunks from several steps travel
// in one drained object; the engine's per-step reads leave it "".
Step string `json:"step,omitempty"`
}
// ErrNoLogs reports an engine that does not serve the log lane yet —
// dagnats#624 unshipped, or a binary that predates it. The page shows
// why the pane is empty instead of an empty pane.
var ErrNoLogs = errors.New("ci: the engine does not serve logs yet (dagnats#624)")
// Logs reads one step's chunks, optionally from the first failure
// (RFC 0009 §logs). follow is the caller's business over SSE; this
// reads the stored chunks one-shot.
func (e *Engine) Logs(runID, step string, afterSeq int, fromFailure bool) ([]LogChunk, error) {
u := e.base + "/runs/" + url.PathEscape(runID) + "/logs?step=" + url.QueryEscape(step)
if afterSeq > 0 {
u += "&after_seq=" + strconv.Itoa(afterSeq)
}
if fromFailure {
u += "&from=failure"
}
req, err := http.NewRequest("GET", u, nil)
if err != nil {
return nil, err
}
resp, err := e.http.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode == http.StatusNotFound {
return nil, ErrNoLogs
}
if resp.StatusCode >= 300 {
return nil, fmt.Errorf("ci: logs %s: %d", runID, resp.StatusCode)
}
body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
if err != nil {
return nil, err
}
var chunks []LogChunk
if err := json.Unmarshal(body, &chunks); err != nil {
// An engine that predates #624 answers the RUN document for
// any sub-path — a JSON object, not an array. That is the
// honest "no logs yet", not a parse failure.
if bytes.HasPrefix(bytes.TrimSpace(body), []byte("{")) {
return nil, ErrNoLogs
}
return nil, err
}
return chunks, nil
}
// Workers lists the engine's connected workers (dagnats#628).
type Worker struct {
WorkerID string `json:"worker_id"`
TaskTypes []string `json:"task_types"`
Language string `json:"language"`
Transport string `json:"transport"`
MaxTasks int `json:"max_tasks"`
Hostname string `json:"hostname,omitempty"`
Version string `json:"version,omitempty"`
}
func (e *Engine) Workers() ([]Worker, error) {
req, err := http.NewRequest("GET", e.base+"/v1/workers", nil)
if err != nil {
return nil, err
}
resp, err := e.http.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode == http.StatusNotFound {
return nil, fmt.Errorf("ci: the engine does not list workers yet (dagnats#628)")
}
if resp.StatusCode >= 300 {
return nil, fmt.Errorf("ci: workers: %d", resp.StatusCode)
}
var out []Worker
body, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
if err != nil {
return nil, err
}
if err := json.Unmarshal(body, &out); err != nil {
var wrapped struct {
Workers []Worker `json:"workers"`
}
if werr := json.Unmarshal(body, &wrapped); werr == nil {
return wrapped.Workers, nil
}
return nil, err
}
return out, nil
}
// Terminal reports whether a run has finished, either way. Anything not
// recognised counts as still running: reporting "done" for a state this
// code has not seen would be the worse mistake.
func (r Run) Terminal() bool {
switch r.Status {
case "completed", "succeeded", "success", "failed", "cancelled", "canceled", "error":
return true
}
return false
}
// Passed reports a successful terminal run.
func (r Run) Passed() bool {
switch r.Status {
case "completed", "succeeded", "success":
return true
}
return false
}
// Register publishes a workflow definition. dagnats is the authority on
// the document's validity, so the raw bytes go up as they were stored
// in the repository.
func (e *Engine) Register(def []byte) (Workflow, error) {
var wf Workflow
if err := json.Unmarshal(def, &wf); err != nil {
return Workflow{}, fmt.Errorf("ci: pipeline is not JSON: %w", err)
}
if wf.Name == "" {
return Workflow{}, fmt.Errorf("ci: pipeline has no name")
}
resp, err := e.post("/workflows", def)
if err != nil {
return Workflow{}, err
}
resp.Body.Close()
if resp.StatusCode >= 300 {
return Workflow{}, fmt.Errorf("ci: register %q: status %d", wf.Name, resp.StatusCode)
}
return wf, nil
}
// Start launches a run, handing the check-in context to the pipeline as
// input: what to build, where it came from, and how to fetch it.
func (e *Engine) Start(workflow string, input map[string]any, labels map[string]string) (string, error) {
doc := map[string]any{"workflow": workflow, "input": input}
if len(labels) > 0 {
doc["labels"] = labels
}
body, err := json.Marshal(doc)
if err != nil {
return "", err
}
resp, err := e.post("/runs", body)
if err != nil {
return "", err
}
defer resp.Body.Close()
if resp.StatusCode >= 300 {
return "", fmt.Errorf("ci: start %q: status %d", workflow, resp.StatusCode)
}
var out struct {
RunID string `json:"run_id"`
}
if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&out); err != nil {
return "", err
}
return out.RunID, nil
}
// Status fetches one run.
func (e *Engine) Status(runID string) (Run, error) {
req, err := http.NewRequest("GET", e.base+"/runs/"+url.PathEscape(runID), nil)
if err != nil {
return Run{}, err
}
resp, err := e.http.Do(req)
if err != nil {
return Run{}, err
}
defer resp.Body.Close()
if resp.StatusCode >= 300 {
return Run{}, fmt.Errorf("ci: status %s: %d", runID, resp.StatusCode)
}
var r Run
if err := json.NewDecoder(io.LimitReader(resp.Body, 4<<20)).Decode(&r); err != nil {
return Run{}, err
}
return r, nil
}
func (e *Engine) post(pathname string, body []byte) (*http.Response, error) {
req, err := http.NewRequest("POST", e.base+pathname, bytes.NewReader(body))
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "application/json")
return e.http.Do(req)
}
// Supersede cancels still-running runs of a workflow that carry every
// given label — the launch path calls it so a new check-in on a branch
// stops the branch's now-stale builds (dagnats#629). Matching is
// verified against each run's own labels before anything is cancelled:
// an engine predating #629 ignores label filters and echoes no labels
// back, and reading that as "cancel everything" would stop other
// branches' builds. Such an engine supersedes nothing, silently.
func (e *Engine) Supersede(workflow string, labels map[string]string) (int, error) {
// The engine's list is a bounded sample filtered AFTER the cap, so a
// small limit can miss the very runs the labels name once the store
// grows. Ask for the ceiling: exact below it, best-effort above.
q := url.Values{"workflow": {workflow}, "limit": {"10000"}}
for k, v := range labels {
q.Add("label", k+"="+v)
}
req, err := http.NewRequest("GET", e.base+"/runs?"+q.Encode(), nil)
if err != nil {
return 0, err
}
resp, err := e.http.Do(req)
if err != nil {
return 0, err
}
defer resp.Body.Close()
if resp.StatusCode >= 300 {
return 0, fmt.Errorf("ci: list runs: %d", resp.StatusCode)
}
var runs []Run
if err := json.NewDecoder(io.LimitReader(resp.Body, 4<<20)).Decode(&runs); err != nil {
return 0, err
}
n := 0
for _, r := range runs {
if r.Terminal() || !labelsCover(r.Labels, labels) {
continue
}
c, err := e.post("/runs/"+url.PathEscape(r.RunID)+"/cancel", nil)
if err != nil {
continue // best-effort: the run may have finished in the race
}
c.Body.Close()
if c.StatusCode < 300 {
n++
}
}
return n, nil
}
// labelsCover reports whether got carries every wanted label.
func labelsCover(got, want map[string]string) bool {
for k, v := range want {
if got[k] != v {
return false
}
}
return true
}
// IsPipeline reports whether a listed path is one of a repository's
// pipeline definitions. Only files directly under ci/ count — a nested
// path is somebody's fixture, not a pipeline, and launching it would be
// surprising.
func IsPipeline(name string) bool {
rest, ok := strings.CutPrefix(name, Dir)
if !ok || rest == "" {
return false
}
return !strings.Contains(rest, "/") && path.Ext(rest) == ".json"
}