Skip to content
quarry Artifact [aba94b088e]
Login

Artifact [aba94b088e]

Check-ins Using Download Hex

Artifact aba94b088eae77ab3711c0858f023b1f5cc728600b7573fc129672367cb4655d:

// 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"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"path"
	"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"`
	Status    string          `json:"status"`
	Steps     map[string]Step `json:"steps"`
	CreatedAt string          `json:"created_at"`
}

// Step is one step's state within a run.
type Step struct {
	Status   string `json:"status"`
	Attempts int    `json:"attempts"`
	Error    string `json:"error,omitempty"`
}

// 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) (string, error) {
	body, err := json.Marshal(map[string]any{"workflow": workflow, "input": input})
	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)
}

// 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"
}