spindle: db: setup pipelines table & helpers

Signed-off-by: Anirudh Oppiliappan <[e-mail]>

spindle/db/db.go

···
 package db
 
-import "database/sql"
+import (
+	"database/sql"
+
+	_ "github.com/mattn/go-sqlite3"
+)
 
 type DB struct {
 	*sql.DB
···
 		);
 
 		create table if not exists pipelines (
-			rkey text not null,
-			pipeline text not null, -- json
-			primary key rkey
+			at_uri text not null,
+			status text not null,
+
+			-- only set if status is 'failed'
+			error text,
+			exit_code integer,
+
+			started_at timestamp not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
+			updated_at timestamp not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
+			finished_at timestamp,
+
+			primary key (at_uri)
 		);
 	`)
 	if err != nil {

spindle/db/pipelines.go

···
 
 import (
 	"fmt"
+	"time"
+
+	"tangled.sh/tangled.sh/core/api/tangled"
+	"tangled.sh/tangled.sh/core/knotserver/notifier"
+)
+
+type PipelineStatus string
+
+var (
+	PipelinePending   PipelineStatus = "pending"
+	PipelineRunning   PipelineStatus = "running"
+	PipelineFailed    PipelineStatus = "failed"
+	PipelineTimeout   PipelineStatus = "timeout"
+	PipelineCancelled PipelineStatus = "cancelled"
+	PipelineSuccess   PipelineStatus = "success"
 )
 
 type Pipeline struct {
-	Rkey         string
-	PipelineJson string
+	Rkey   string         `json:"rkey"`
+	Knot   string         `json:"knot"`
+	Status PipelineStatus `json:"status"`
+
+	// only if Failed
+	Error    string `json:"error"`
+	ExitCode int    `json:"exit_code"`
+
+	StartedAt  time.Time `json:"started_at"`
+	UpdatedAt  time.Time `json:"updated_at"`
+	FinishedAt time.Time `json:"finished_at"`
 }
 
-func (d *DB) InsertPipeline(pipeline Pipeline) error {
-	_, err := d.Exec(
-		`insert into pipelines (rkey, nsid, event) values (?, ?, ?)`,
-		pipeline.Rkey,
-		pipeline.PipelineJson,
-	)
+func (p Pipeline) AsRecord() *tangled.PipelineStatus {
+	exitCode64 := int64(p.ExitCode)
+	finishedAt := p.FinishedAt.String()
+
+	return &tangled.PipelineStatus{
+		Pipeline: fmt.Sprintf("at://%s/%s", p.Knot, p.Rkey),
+		Status:   string(p.Status),
+
+		ExitCode: &exitCode64,
+		Error:    &p.Error,
+
+		StartedAt:  p.StartedAt.String(),
+		UpdatedAt:  p.UpdatedAt.String(),
+		FinishedAt: &finishedAt,
+	}
+}
 
-	return err
+func pipelineAtUri(rkey, knot string) string {
+	return fmt.Sprintf("at://%s/did:web:%s/%s", tangled.PipelineStatusNSID, knot, rkey)
 }
 
-func (d *DB) GetPipeline(rkey, cursor string) (Pipeline, error) {
-	whereClause := "where rkey = ?"
-	args := []any{rkey}
+func (db *DB) CreatePipeline(rkey, knot string, n *notifier.Notifier) error {
+	_, err := db.Exec(`
+		insert into pipelines (at_uri, status)
+		values (?, ?)
+	`, pipelineAtUri(rkey, knot), PipelinePending)
 
-	if cursor != "" {
-		whereClause += " and rkey > ?"
-		args = append(args, cursor)
+	if err != nil {
+		return err
 	}
+	n.NotifyAll()
+	return nil
+}
 
-	query := fmt.Sprintf(`
-		select rkey, pipeline
-		from pipelines
-		%s
-		limit 1
-	`, whereClause)
+func (db *DB) MarkPipelineRunning(rkey, knot string, n *notifier.Notifier) error {
+	_, err := db.Exec(`
+			update pipelines
+			set status = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
+			where at_uri = ?
+		`, PipelineRunning, pipelineAtUri(rkey, knot))
 
-	row := d.QueryRow(query, args...)
+	if err != nil {
+		return err
+	}
+	n.NotifyAll()
+	return nil
+}
+
+func (db *DB) MarkPipelineFailed(rkey, knot string, exitCode int, errorMsg string, n *notifier.Notifier) error {
+	_, err := db.Exec(`
+		update pipelines
+		set status = ?,
+		    exit_code = ?,
+		    error = ?,
+		    updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now'),
+		    finished_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
+		where at_uri = ?
+	`, PipelineFailed, exitCode, errorMsg, pipelineAtUri(rkey, knot))
+	if err != nil {
+		return err
+	}
+	n.NotifyAll()
+	return nil
+}
+
+func (db *DB) MarkPipelineTimeout(rkey, knot string, n *notifier.Notifier) error {
+	_, err := db.Exec(`
+			update pipelines
+			set status = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
+			where at_uri = ?
+		`, PipelineTimeout, pipelineAtUri(rkey, knot))
+	if err != nil {
+		return err
+	}
+	n.NotifyAll()
+	return nil
+}
+
+func (db *DB) MarkPipelineSuccess(rkey, knot string, n *notifier.Notifier) error {
+	_, err := db.Exec(`
+			update pipelines
+			set status = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now'),
+			finished_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
+			where at_uri = ?
+		`, PipelineSuccess, pipelineAtUri(rkey, knot))
 
-	var p Pipeline
-	err := row.Scan(&p.Rkey, &p.PipelineJson)
 	if err != nil {
-		return Pipeline{}, err
+		return err
 	}
+	n.NotifyAll()
+	return nil
+}
 
-	return p, nil
+func (db *DB) GetPipeline(rkey, knot string) (Pipeline, error) {
+	var p Pipeline
+	err := db.QueryRow(`
+		select rkey, status, error, exit_code, started_at, updated_at, finished_at
+		from pipelines
+		where at_uri = ?
+	`, pipelineAtUri(rkey, knot)).Scan(&p.Rkey, &p.Status, &p.Error, &p.ExitCode, &p.StartedAt, &p.UpdatedAt, &p.FinishedAt)
+	return p, err
 }
 
-func (d *DB) GetPipelines(cursor string) ([]Pipeline, error) {
+func (db *DB) GetPipelines(cursor string) ([]Pipeline, error) {
 	whereClause := ""
 	args := []any{}
 	if cursor != "" {
···
 	}
 
 	query := fmt.Sprintf(`
-		select rkey, nsid, pipeline
+		select rkey, status, error, exit_code, started_at, updated_at, finished_at
 		from pipelines
 		%s
 		order by rkey asc
 		limit 100
 	`, whereClause)
 
-	rows, err := d.Query(query, args...)
+	rows, err := db.Query(query, args...)
 	if err != nil {
 		return nil, err
 	}
 	defer rows.Close()
 
-	var evts []Pipeline
+	var pipelines []Pipeline
 	for rows.Next() {
-		var ev Pipeline
-		rows.Scan(&ev.Rkey, &ev.PipelineJson)
-		evts = append(evts, ev)
+		var p Pipeline
+		rows.Scan(&p.Rkey, &p.Status, &p.Error, &p.ExitCode, &p.StartedAt, &p.UpdatedAt, &p.FinishedAt)
+		pipelines = append(pipelines, p)
 	}
 
 	if err := rows.Err(); err != nil {
 		return nil, err
 	}
 
-	return evts, nil
+	return pipelines, nil
 }