Artifact [191650af9b]

Artifact 191650af9ba314215b4c73481f78952c6d746f604debb8abc1c29f1bc839aba8:

• File internal/oltp/ci.go — part of check-in [0502a93834] at 2026-08-17 13:02:21 on branch trunk — ci: the trigger half of v3 — dagnats is the engine

  0001 §6 splits this narrowly: dagnats runs pipelines, quarry notices a check-in and shows the result. quarry defines nothing, schedules nothing, retries nothing.

  A pipeline is TRACKED CONTENT — ci/*.json in the repository, dagnats' native format — read at the check-in under test, so the pipeline for a historical check-in is the pipeline that ran and there is no forge-side store to drift. Only files directly under ci/ count; a nested path is somebody's fixture.

  Two behaviours worth naming. First sight of a repository takes the tip as a watermark and launches NOTHING: running a pipeline per historical check-in would be a denial of service against the engine and would say nothing about code nobody just changed — on an imported corpus that is tens of thousands of runs. And an unrecognised run status counts as still running, because claiming a run finished when it has not is how a verdict gets written against a build still going.

  Verified against the live dagnats for the wire shapes; the trigger itself is covered by tests. (user: dmestas, size: 3895) [annotate] [blame] [check-ins using]

// 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,
	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
);`

// CIRun is one launched pipeline.
type CIRun struct {
	RunID    string `json:"runId"`
	Repo     string `json:"repo"`
	Checkin  string `json:"checkin"`
	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, workflow, pipeline, status, started)
		VALUES (?, ?, ?, ?, ?, ?, ?)`,
		r.RunID, r.Repo, r.Checkin, 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
}

// 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, 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.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:])
}