wip: adding database ctx to pipelines

api/build/create.go

@@ -232,7 +232,7 @@ func CreateBuild(c *gin.Context) {
 	)
 
 	// send API call to attempt to capture the pipeline
-	pipeline, err = database.FromContext(c).GetPipelineForRepo(input.GetCommit(), r)
+	pipeline, err = database.FromContext(c).GetPipelineForRepo(ctx, input.GetCommit(), r)
 	if err != nil { // assume the pipeline doesn't exist in the database yet
 		// send API call to capture the pipeline configuration file
 		config, err = scm.FromContext(c).ConfigBackoff(u, r, input.GetCommit())
@@ -309,7 +309,7 @@ func CreateBuild(c *gin.Context) {
 		pipeline.SetRef(input.GetRef())
 
 		// send API call to create the pipeline
-		pipeline, err = database.FromContext(c).CreatePipeline(pipeline)
+		pipeline, err = database.FromContext(c).CreatePipeline(ctx, pipeline)
 		if err != nil {
 			retErr := fmt.Errorf("unable to create new build: failed to create pipeline for %s: %w", r.GetFullName(), err)
 

api/build/restart.go

@@ -223,7 +223,7 @@ func RestartBuild(c *gin.Context) {
 	)
 
 	// send API call to attempt to capture the pipeline
-	pipeline, err = database.FromContext(c).GetPipelineForRepo(b.GetCommit(), r)
+	pipeline, err = database.FromContext(c).GetPipelineForRepo(ctx, b.GetCommit(), r)
 	if err != nil { // assume the pipeline doesn't exist in the database yet (before pipeline support was added)
 		// send API call to capture the pipeline configuration file
 		config, err = scm.FromContext(c).ConfigBackoff(u, r, b.GetCommit())
@@ -300,7 +300,7 @@ func RestartBuild(c *gin.Context) {
 		pipeline.SetRef(b.GetRef())
 
 		// send API call to create the pipeline
-		pipeline, err = database.FromContext(c).CreatePipeline(pipeline)
+		pipeline, err = database.FromContext(c).CreatePipeline(ctx, pipeline)
 		if err != nil {
 			retErr := fmt.Errorf("unable to create pipeline for %s: %w", r.GetFullName(), err)
 

api/pipeline/create.go

@@ -70,6 +70,7 @@ func CreatePipeline(c *gin.Context) {
 	o := org.Retrieve(c)
 	r := repo.Retrieve(c)
 	u := user.Retrieve(c)
+	ctx := c.Request.Context()
 
 	// update engine logger with API metadata
 	//
@@ -98,7 +99,7 @@ func CreatePipeline(c *gin.Context) {
 	input.SetRepoID(r.GetID())
 
 	// send API call to create the pipeline
-	p, err := database.FromContext(c).CreatePipeline(input)
+	p, err := database.FromContext(c).CreatePipeline(ctx, input)
 	if err != nil {
 		retErr := fmt.Errorf("unable to create pipeline %s/%s: %w", r.GetFullName(), input.GetCommit(), err)
 

api/pipeline/delete.go

@@ -65,6 +65,7 @@ func DeletePipeline(c *gin.Context) {
 	p := pipeline.Retrieve(c)
 	r := repo.Retrieve(c)
 	u := user.Retrieve(c)
+	ctx := c.Request.Context()
 
 	entry := fmt.Sprintf("%s/%s", r.GetFullName(), p.GetCommit())
 
@@ -79,7 +80,7 @@ func DeletePipeline(c *gin.Context) {
 	}).Infof("deleting pipeline %s", entry)
 
 	// send API call to remove the build
-	err := database.FromContext(c).DeletePipeline(p)
+	err := database.FromContext(c).DeletePipeline(ctx, p)
 	if err != nil {
 		retErr := fmt.Errorf("unable to delete pipeline %s: %w", entry, err)
 

api/pipeline/list.go

@@ -80,6 +80,7 @@ func ListPipelines(c *gin.Context) {
 	o := org.Retrieve(c)
 	r := repo.Retrieve(c)
 	u := user.Retrieve(c)
+	ctx := c.Request.Context()
 
 	// update engine logger with API metadata
 	//
@@ -117,7 +118,7 @@ func ListPipelines(c *gin.Context) {
 	//nolint:gomnd // ignore magic number
 	perPage = util.MaxInt(1, util.MinInt(100, perPage))
 
-	p, t, err := database.FromContext(c).ListPipelinesForRepo(r, page, perPage)
+	p, t, err := database.FromContext(c).ListPipelinesForRepo(ctx, r, page, perPage)
 	if err != nil {
 		retErr := fmt.Errorf("unable to list pipelines for repo %s: %w", r.GetFullName(), err)
 

api/pipeline/update.go

@@ -72,6 +72,7 @@ func UpdatePipeline(c *gin.Context) {
 	p := pipeline.Retrieve(c)
 	r := repo.Retrieve(c)
 	u := user.Retrieve(c)
+	ctx := c.Request.Context()
 
 	entry := fmt.Sprintf("%s/%s", r.GetFullName(), p.GetCommit())
 
@@ -170,7 +171,7 @@ func UpdatePipeline(c *gin.Context) {
 	}
 
 	// send API call to update the pipeline
-	p, err = database.FromContext(c).UpdatePipeline(p)
+	p, err = database.FromContext(c).UpdatePipeline(ctx, p)
 	if err != nil {
 		retErr := fmt.Errorf("unable to update pipeline %s: %w", entry, err)
 

api/webhook/post.go

@@ -426,7 +426,7 @@ func PostWebhook(c *gin.Context) {
 		}
 
 		// send API call to attempt to capture the pipeline
-		pipeline, err = database.FromContext(c).GetPipelineForRepo(b.GetCommit(), repo)
+		pipeline, err = database.FromContext(c).GetPipelineForRepo(ctx, b.GetCommit(), repo)
 		if err != nil { // assume the pipeline doesn't exist in the database yet
 			// send API call to capture the pipeline configuration file
 			config, err = scm.FromContext(c).ConfigBackoff(u, repo, b.GetCommit())
@@ -561,7 +561,7 @@ func PostWebhook(c *gin.Context) {
 			pipeline.SetRef(b.GetRef())
 
 			// send API call to create the pipeline
-			pipeline, err = database.FromContext(c).CreatePipeline(pipeline)
+			pipeline, err = database.FromContext(c).CreatePipeline(ctx, pipeline)
 			if err != nil {
 				retErr := fmt.Errorf("%s: failed to create pipeline for %s: %w", baseErr, repo.GetFullName(), err)
 

cmd/vela-server/schedule.go

@@ -237,7 +237,7 @@ func processSchedule(s *library.Schedule, compiler compiler.Engine, database dat
 		}
 
 		// send API call to attempt to capture the pipeline
-		pipeline, err = database.GetPipelineForRepo(b.GetCommit(), r)
+		pipeline, err = database.GetPipelineForRepo(context.TODO(), b.GetCommit(), r)
 		if err != nil { // assume the pipeline doesn't exist in the database yet
 			// send API call to capture the pipeline configuration file
 			config, err = scm.ConfigBackoff(u, r, b.GetCommit())
@@ -326,7 +326,7 @@ func processSchedule(s *library.Schedule, compiler compiler.Engine, database dat
 			pipeline.SetRef(b.GetRef())
 
 			// send API call to create the pipeline
-			pipeline, err = database.CreatePipeline(pipeline)
+			pipeline, err = database.CreatePipeline(context.TODO(), pipeline)
 			if err != nil {
 				err = fmt.Errorf("failed to create pipeline for %s: %w", r.GetFullName(), err)
 

database/database.go

@@ -5,6 +5,7 @@
 package database
 
 import (
+	"context"
 	"fmt"
 	"time"
 
@@ -54,6 +55,8 @@ type (
 		client *gorm.DB
 		// engine configuration settings used in database functions
 		config *config
+		// engine context used in database functions
+		ctx context.Context
 		// sirupsen/logrus logger used in database functions
 		logger *logrus.Entry
 
@@ -85,6 +88,7 @@ func New(opts ...EngineOpt) (Interface, error) {
 	e.client = new(gorm.DB)
 	e.config = new(config)
 	e.logger = new(logrus.Entry)
+	e.ctx = context.TODO()
 
 	// apply all provided configuration options
 	for _, opt := range opts {
@@ -143,7 +147,7 @@ func New(opts ...EngineOpt) (Interface, error) {
 	}
 
 	// create database agnostic engines for resources
-	err = e.NewResources()
+	err = e.NewResources(e.ctx)
 	if err != nil {
 		return nil, err
 	}

database/integration_test.go

@@ -664,15 +664,15 @@ func testPipelines(t *testing.T, db Interface, resources *Resources) {
 
 	// create the pipelines
 	for _, pipeline := range resources.Pipelines {
-		_, err := db.CreatePipeline(pipeline)
+		_, err := db.CreatePipeline(context.TODO(), pipeline)
 		if err != nil {
 			t.Errorf("unable to create pipeline %d: %v", pipeline.GetID(), err)
 		}
 	}
 	methods["CreatePipeline"] = true
 
 	// count the pipelines
-	count, err := db.CountPipelines()
+	count, err := db.CountPipelines(context.TODO())
 	if err != nil {
 		t.Errorf("unable to count pipelines: %v", err)
 	}
@@ -682,7 +682,7 @@ func testPipelines(t *testing.T, db Interface, resources *Resources) {
 	methods["CountPipelines"] = true
 
 	// count the pipelines for a repo
-	count, err = db.CountPipelinesForRepo(resources.Repos[0])
+	count, err = db.CountPipelinesForRepo(context.TODO(), resources.Repos[0])
 	if err != nil {
 		t.Errorf("unable to count pipelines for repo %d: %v", resources.Repos[0].GetID(), err)
 	}
@@ -692,7 +692,7 @@ func testPipelines(t *testing.T, db Interface, resources *Resources) {
 	methods["CountPipelinesForRepo"] = true
 
 	// list the pipelines
-	list, err := db.ListPipelines()
+	list, err := db.ListPipelines(context.TODO())
 	if err != nil {
 		t.Errorf("unable to list pipelines: %v", err)
 	}
@@ -702,7 +702,7 @@ func testPipelines(t *testing.T, db Interface, resources *Resources) {
 	methods["ListPipelines"] = true
 
 	// list the pipelines for a repo
-	list, count, err = db.ListPipelinesForRepo(resources.Repos[0], 1, 10)
+	list, count, err = db.ListPipelinesForRepo(context.TODO(), resources.Repos[0], 1, 10)
 	if err != nil {
 		t.Errorf("unable to list pipelines for repo %d: %v", resources.Repos[0].GetID(), err)
 	}
@@ -717,7 +717,7 @@ func testPipelines(t *testing.T, db Interface, resources *Resources) {
 	// lookup the pipelines by name
 	for _, pipeline := range resources.Pipelines {
 		repo := resources.Repos[pipeline.GetRepoID()-1]
-		got, err := db.GetPipelineForRepo(pipeline.GetCommit(), repo)
+		got, err := db.GetPipelineForRepo(context.TODO(), pipeline.GetCommit(), repo)
 		if err != nil {
 			t.Errorf("unable to get pipeline %d for repo %d: %v", pipeline.GetID(), repo.GetID(), err)
 		}
@@ -730,13 +730,13 @@ func testPipelines(t *testing.T, db Interface, resources *Resources) {
 	// update the pipelines
 	for _, pipeline := range resources.Pipelines {
 		pipeline.SetVersion("2")
-		_, err = db.UpdatePipeline(pipeline)
+		_, err = db.UpdatePipeline(context.TODO(), pipeline)
 		if err != nil {
 			t.Errorf("unable to update pipeline %d: %v", pipeline.GetID(), err)
 		}
 
 		// lookup the pipeline by ID
-		got, err := db.GetPipeline(pipeline.GetID())
+		got, err := db.GetPipeline(context.TODO(), pipeline.GetID())
 		if err != nil {
 			t.Errorf("unable to get pipeline %d by ID: %v", pipeline.GetID(), err)
 		}
@@ -749,7 +749,7 @@ func testPipelines(t *testing.T, db Interface, resources *Resources) {
 
 	// delete the pipelines
 	for _, pipeline := range resources.Pipelines {
-		err = db.DeletePipeline(pipeline)
+		err = db.DeletePipeline(context.TODO(), pipeline)
 		if err != nil {
 			t.Errorf("unable to delete pipeline %d: %v", pipeline.GetID(), err)
 		}

database/pipeline/count_repo_test.go

@@ -5,6 +5,7 @@
 package pipeline
 
 import (
+	"context"
 	"reflect"
 	"testing"
 
@@ -42,12 +43,12 @@ func TestPipeline_Engine_CountPipelinesForRepo(t *testing.T) {
 	_sqlite := testSqlite(t)
 	defer func() { _sql, _ := _sqlite.client.DB(); _sql.Close() }()
 
-	_, err := _sqlite.CreatePipeline(_pipelineOne)
+	_, err := _sqlite.CreatePipeline(context.TODO(), _pipelineOne)
 	if err != nil {
 		t.Errorf("unable to create test pipeline for sqlite: %v", err)
 	}
 
-	_, err = _sqlite.CreatePipeline(_pipelineTwo)
+	_, err = _sqlite.CreatePipeline(context.TODO(), _pipelineTwo)
 	if err != nil {
 		t.Errorf("unable to create test pipeline for sqlite: %v", err)
 	}
@@ -76,7 +77,7 @@ func TestPipeline_Engine_CountPipelinesForRepo(t *testing.T) {
 	// run tests
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			got, err := test.database.CountPipelinesForRepo(&library.Repo{ID: _pipelineOne.RepoID})
+			got, err := test.database.CountPipelinesForRepo(context.TODO(), &library.Repo{ID: _pipelineOne.RepoID})
 
 			if test.failure {
 				if err == nil {

database/pipeline/count_test.go

@@ -5,6 +5,7 @@
 package pipeline
 
 import (
+	"context"
 	"reflect"
 	"testing"
 
@@ -41,12 +42,12 @@ func TestPipeline_Engine_CountPipelines(t *testing.T) {
 	_sqlite := testSqlite(t)
 	defer func() { _sql, _ := _sqlite.client.DB(); _sql.Close() }()
 
-	_, err := _sqlite.CreatePipeline(_pipelineOne)
+	_, err := _sqlite.CreatePipeline(context.TODO(), _pipelineOne)
 	if err != nil {
 		t.Errorf("unable to create test pipeline for sqlite: %v", err)
 	}
 
-	_, err = _sqlite.CreatePipeline(_pipelineTwo)
+	_, err = _sqlite.CreatePipeline(context.TODO(), _pipelineTwo)
 	if err != nil {
 		t.Errorf("unable to create test pipeline for sqlite: %v", err)
 	}
@@ -75,7 +76,7 @@ func TestPipeline_Engine_CountPipelines(t *testing.T) {
 	// run tests
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			got, err := test.database.CountPipelines()
+			got, err := test.database.CountPipelines(context.TODO())
 
 			if test.failure {
 				if err == nil {

database/pipeline/create_test.go

@@ -5,6 +5,7 @@
 package pipeline
 
 import (
+	"context"
 	"reflect"
 	"testing"
 
@@ -59,7 +60,7 @@ VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15) RETURNING "id"`).
 	// run tests
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			got, err := test.database.CreatePipeline(_pipeline)
+			got, err := test.database.CreatePipeline(context.TODO(), _pipeline)
 
 			if test.failure {
 				if err == nil {

database/pipeline/delete_test.go

@@ -5,6 +5,7 @@
 package pipeline
 
 import (
+	"context"
 	"testing"
 
 	"github.com/DATA-DOG/go-sqlmock"
@@ -31,7 +32,7 @@ func TestPipeline_Engine_DeletePipeline(t *testing.T) {
 	_sqlite := testSqlite(t)
 	defer func() { _sql, _ := _sqlite.client.DB(); _sql.Close() }()
 
-	_, err := _sqlite.CreatePipeline(_pipeline)
+	_, err := _sqlite.CreatePipeline(context.TODO(), _pipeline)
 	if err != nil {
 		t.Errorf("unable to create test pipeline for sqlite: %v", err)
 	}
@@ -57,7 +58,7 @@ func TestPipeline_Engine_DeletePipeline(t *testing.T) {
 	// run tests
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			err = test.database.DeletePipeline(_pipeline)
+			err = test.database.DeletePipeline(context.TODO(), _pipeline)
 
 			if test.failure {
 				if err == nil {

database/pipeline/get_repo_test.go

@@ -5,6 +5,7 @@
 package pipeline
 
 import (
+	"context"
 	"reflect"
 	"testing"
 
@@ -37,7 +38,7 @@ func TestPipeline_Engine_GetPipelineForRepo(t *testing.T) {
 	_sqlite := testSqlite(t)
 	defer func() { _sql, _ := _sqlite.client.DB(); _sql.Close() }()
 
-	_, err := _sqlite.CreatePipeline(_pipeline)
+	_, err := _sqlite.CreatePipeline(context.TODO(), _pipeline)
 	if err != nil {
 		t.Errorf("unable to create test pipeline for sqlite: %v", err)
 	}
@@ -66,7 +67,7 @@ func TestPipeline_Engine_GetPipelineForRepo(t *testing.T) {
 	// run tests
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			got, err := test.database.GetPipelineForRepo("48afb5bdc41ad69bf22588491333f7cf71135163", &library.Repo{ID: _pipeline.RepoID})
+			got, err := test.database.GetPipelineForRepo(context.TODO(), "48afb5bdc41ad69bf22588491333f7cf71135163", &library.Repo{ID: _pipeline.RepoID})
 
 			if test.failure {
 				if err == nil {