Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240package db
import ( "context" "database/sql" "encoding/json" "errors" "strconv" "strings"
"tangled.org/core/orm" "tangled.org/core/spindle/models")
func (d *DB) QueryPipelines(ctx context.Context, repoDID string, commits []string, cursor string, kinds []string, limit int) ([]*models.PipelineRecord, string, int64, error) { if limit <= 0 { limit = 30 }
filters := []orm.Filter{orm.FilterEq("repo_did", repoDID)} if len(commits) > 0 { filters = append(filters, orm.FilterIn("commit_sha", commits)) } if len(kinds) > 0 { filters = append(filters, orm.FilterIn("kind", kinds)) }
var conditions []string var args []any for _, filter := range filters { conditions = append(conditions, filter.Condition()) args = append(args, filter.Arg()...) } whereClause := " where " + strings.Join(conditions, " and ")
// later pages skip the count since the cursor carries the total var total int64 counted := false cursorID, hasCursor := int64(0), false if cursor != "" { idText, totalText, _ := strings.Cut(cursor, ":") if value, err := strconv.ParseInt(idText, 10, 64); err == nil { cursorID, hasCursor = value, true if carried, err := strconv.ParseInt(totalText, 10, 64); err == nil { total, counted = carried, true } } } if !counted { if err := d.QueryRowContext(ctx, `select count(*) from pipelines`+whereClause, args...).Scan(&total); err != nil { return nil, "", 0, err } } if hasCursor { filter := orm.FilterLt("id", cursorID) conditions = append(conditions, filter.Condition()) args = append(args, filter.Arg()...) whereClause = " where " + strings.Join(conditions, " and ") } args = append(args, limit)
rows, err := d.QueryContext(ctx, `select id, payload from pipelines`+whereClause+` order by id desc limit ?`, args...) if err != nil { return nil, "", 0, err } defer rows.Close()
records := make([]*models.PipelineRecord, 0, limit) var lastID int64 var scanned int for rows.Next() { var id int64 var payload string if err := rows.Scan(&id, &payload); err != nil { return nil, "", 0, err } lastID = id scanned++ var record models.PipelineRecord if json.Unmarshal([]byte(payload), &record) != nil { continue } records = append(records, &record) } if err := rows.Err(); err != nil { return nil, "", 0, err } if err := d.applyStatuses(ctx, records); err != nil { return nil, "", 0, err }
nextCursor := "" if scanned == limit { nextCursor = strconv.FormatInt(lastID, 10) + ":" + strconv.FormatInt(total, 10) } return records, nextCursor, total, nil}
func (d *DB) GetPipeline(ctx context.Context, id models.PipelineId) (*models.PipelineRecord, error) { var payload string if err := d.QueryRowContext(ctx, `select payload from pipelines where pipeline_id = ?`, id, ).Scan(&payload); err != nil { return nil, err }
var record models.PipelineRecord if err := json.Unmarshal([]byte(payload), &record); err != nil { return nil, err } if err := d.applyStatuses(ctx, []*models.PipelineRecord{&record}); err != nil { return nil, err } return &record, nil}
func (d *DB) CreatePipeline(record *models.PipelineRecord) error { return insertPipeline(d, record)}
func insertPipeline(q DBTX, record *models.PipelineRecord) error { payload, err := json.Marshal(record) if err != nil { return err } var ref, oldSha *string if push := record.Trigger.Push; push != nil { ref, oldSha = &push.Ref, &push.OldCommit } _, err = q.Exec( `insert into pipelines (pipeline_id, repo_did, commit_sha, kind, payload, ref, old_sha) values (?, ?, ?, ?, ?, ?, ?)`, record.ID, record.RepoDID, record.Commit, record.Trigger.Kind, string(payload), ref, oldSha, ) return err}
// a replayed notify lands here so the pipeline doesn't run twicefunc (d *DB) PushPipeline(repoDid, ref, oldSha, newSha string) (id models.PipelineId, found bool, err error) { err = d.QueryRow( `select pipeline_id from pipelines where kind = 'push' and repo_did = ? and commit_sha = ? and ref = ? and old_sha = ? limit 1`, repoDid, newSha, ref, oldSha, ).Scan(&id) if errors.Is(err, sql.ErrNoRows) { return "", false, nil } return id, err == nil, err}
func (d *DB) applyStatuses(ctx context.Context, pipelines []*models.PipelineRecord) error { if len(pipelines) == 0 { return nil }
pipelineIDs := make([]string, 0, len(pipelines)) for _, pipeline := range pipelines { pipelineIDs = append(pipelineIDs, string(pipeline.ID)) }
statuses, err := d.workflowStatuses(ctx, pipelineIDs) if err != nil { return err }
for _, pipeline := range pipelines { for _, workflow := range pipeline.Workflows { if workflow == nil { continue } status, ok := statuses[wfKey{string(pipeline.ID), workflow.Name}] if !ok { continue } workflow.Status = status.Status workflow.Error = status.Error workflow.StartedAt = status.StartedAt workflow.FinishedAt = status.FinishedAt } } return nil}
type wfKey struct { PipelineID string Workflow string}
type wfStatus struct { Status string Error *string StartedAt *string FinishedAt *string}
func (d *DB) workflowStatuses(ctx context.Context, pipelineIDs []string) (map[wfKey]wfStatus, error) { filter := orm.FilterIn("pipeline_id", pipelineIDs) rows, err := d.QueryContext(ctx, `select pipeline_id, workflow, status, error, created_at from workflow_statuses where `+filter.Condition()+` order by id asc`, filter.Arg()...) if err != nil { return nil, err } defer rows.Close()
out := make(map[wfKey]wfStatus) for rows.Next() { var pipelineID, wfName, status, createdAt string var wfError *string if err := rows.Scan(&pipelineID, &wfName, &status, &wfError, &createdAt); err != nil { return nil, err }
k := wfKey{pipelineID, wfName} st := out[k]
// rows arrive in insertion order, so the last one wins for the current status st.Status = status st.Error = wfError
switch kind := models.StatusKind(status); { case kind == models.StatusKindRunning: // the first running row is when the workflow actually started if st.StartedAt == nil { at := createdAt st.StartedAt = &at } case kind.IsFinish(): at := createdAt st.FinishedAt = &at }
out[k] = st } return out, rows.Err()}