diff --git a/spindle/db/db.go b/spindle/db/db.go index a2292e2..630a0b0 100644 --- a/spindle/db/db.go +++ b/spindle/db/db.go @@ -1,6 +1,10 @@ package db -import "database/sql" +import ( + "database/sql" + + _ "github.com/mattn/go-sqlite3" +) type DB struct { *sql.DB @@ -27,9 +31,18 @@ func Make(dbPath string) (*DB, error) { ); 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 { diff --git a/spindle/db/pipelines.go b/spindle/db/pipelines.go index 8e4f31b..df447c2 100644 --- a/spindle/db/pipelines.go +++ b/spindle/db/pipelines.go @@ -2,51 +2,141 @@ package db 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 +} - var p Pipeline - err := row.Scan(&p.Rkey, &p.PipelineJson) +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 Pipeline{}, err + return err } + n.NotifyAll() + return nil +} - return p, 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)) + + if err != nil { + return err + } + n.NotifyAll() + return 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 != "" { @@ -55,29 +145,29 @@ func (d *DB) GetPipelines(cursor string) ([]Pipeline, error) { } 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 }