// ci.go records which dagnats run belongs to which check-in.
//
// The run itself lives in dagnats — quarry does not duplicate its event
// log. What quarry must remember is the mapping, because dagnats knows
// only that some workflow ran with some input, and the forge is what
// knows that input was a check-in in a repository (0001 §6).
package oltp
import "time"
const ciSchema = `
CREATE TABLE IF NOT EXISTS ci_run (
run_id TEXT PRIMARY KEY, -- dagnats run id
repo TEXT NOT NULL,
checkin TEXT NOT NULL,
pair TEXT NOT NULL DEFAULT '', -- premerge pair, RFC 0009
workflow TEXT NOT NULL,
pipeline TEXT NOT NULL, -- the ci/*.json path this came from
status TEXT NOT NULL, -- last status seen; dagnats remains authoritative
started TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS ci_run_by_checkin ON ci_run (repo, checkin);
CREATE INDEX IF NOT EXISTS ci_run_by_repo ON ci_run (repo, started DESC);
-- The watermark is how quarry avoids launching a run per restart: it
-- remembers the newest check-in it has already considered, per repo.
CREATE TABLE IF NOT EXISTS ci_watermark (
repo TEXT PRIMARY KEY,
checkin TEXT NOT NULL,
seen TEXT NOT NULL
);`
// ciMigrations widen older stores (RFC 0009 §premerge): a run may be
// about a pair of tips rather than one check-in, and a branch rule may
// ask for premerge testing.
var ciMigrations = []string{
`ALTER TABLE ci_run ADD COLUMN pair TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE branch_required_checks ADD COLUMN mode TEXT NOT NULL DEFAULT ''`,
// After the column exists, never beside CREATE TABLE: on a store
// that already has ci_run the CREATE is a no-op, so an index over
// `pair` in the schema block names a column this ALTER has not
// added yet and the open fails outright.
`CREATE INDEX IF NOT EXISTS ci_run_by_pair ON ci_run (repo, pair)`,
}
// CIRun is one launched pipeline.
type CIRun struct {
RunID string `json:"runId"`
Repo string `json:"repo"`
Checkin string `json:"checkin"`
// Pair marks a premerge run: the verdict belongs to this pair of
// tips, never to a check-in (RFC 0009 §premerge). Empty for tip runs.
Pair string `json:"pair,omitempty"`
Workflow string `json:"workflow"`
Pipeline string `json:"pipeline"`
Status string `json:"status"`
Started string `json:"started"`
}
// AddCIRun records a launch.
func (d *DB) AddCIRun(r CIRun) error {
_, err := d.db.Exec(`INSERT OR REPLACE INTO ci_run
(run_id, repo, checkin, pair, workflow, pipeline, status, started)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
r.RunID, r.Repo, r.Checkin, r.Pair, r.Workflow, r.Pipeline, r.Status,
time.Now().UTC().Format(time.RFC3339))
return err
}
// SetCIStatus updates the cached status of a run.
func (d *DB) SetCIStatus(runID, status string) error {
_, err := d.db.Exec(`UPDATE ci_run SET status = ? WHERE run_id = ?`, status, runID)
return err
}
// CIRunsForPair lists a premerge pair's runs, newest first.
func (d *DB) CIRunsForPair(repo, pair string) ([]CIRun, error) {
return d.ciRuns(`WHERE repo = ? AND pair = ? ORDER BY started DESC`, repo, pair)
}
// CIRunsFor lists the runs launched for one check-in, newest first.
func (d *DB) CIRunsFor(repo, checkin string) ([]CIRun, error) {
return d.ciRuns(`WHERE repo = ? AND checkin = ? ORDER BY started DESC`, repo, checkin)
}
// CIRuns lists a repository's recent runs.
func (d *DB) CIRuns(repo string, limit int) ([]CIRun, error) {
if limit <= 0 {
limit = 50
}
return d.ciRuns(`WHERE repo = ? ORDER BY started DESC LIMIT `+itoa(limit), repo)
}
// UnfinishedCIRuns lists runs whose cached status is not terminal, so a
// poller knows what to ask dagnats about.
func (d *DB) UnfinishedCIRuns() ([]CIRun, error) {
return d.ciRuns(`WHERE status NOT IN
('completed','succeeded','success','failed','cancelled','canceled','error')`)
}
func (d *DB) ciRuns(where string, args ...any) ([]CIRun, error) {
rows, err := d.db.Query(`SELECT run_id, repo, checkin, pair, workflow, pipeline, status, started
FROM ci_run `+where, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var out []CIRun
for rows.Next() {
var r CIRun
if err := rows.Scan(&r.RunID, &r.Repo, &r.Checkin, &r.Pair, &r.Workflow,
&r.Pipeline, &r.Status, &r.Started); err != nil {
return nil, err
}
out = append(out, r)
}
return out, rows.Err()
}
// CIWatermark returns the newest check-in already considered for a repo.
func (d *DB) CIWatermark(repo string) (string, bool) {
var c string
err := d.db.QueryRow(`SELECT checkin FROM ci_watermark WHERE repo = ?`, repo).Scan(&c)
return c, err == nil
}
// SetCIWatermark records how far CI has looked.
func (d *DB) SetCIWatermark(repo, checkin string) error {
_, err := d.db.Exec(`INSERT OR REPLACE INTO ci_watermark (repo, checkin, seen)
VALUES (?, ?, ?)`, repo, checkin, time.Now().UTC().Format(time.RFC3339))
return err
}
func itoa(n int) string {
if n == 0 {
return "0"
}
var b [20]byte
i := len(b)
for n > 0 {
i--
b[i] = byte('0' + n%10)
n /= 10
}
return string(b[i:])
}