fix(database)!: use transactions tied to repo counters for hook and build to avoid collisions

api/build/compile_publish.go

@@ -186,8 +186,6 @@ func CompileAndPublish(
 		pipeline *types.Pipeline
 		// variable to store the pipeline type for the repository
 		pipelineType = r.GetPipelineType()
-		// variable to store updated repository record
-		repo *types.Repo
 	)
 
 	// implement a loop to process asynchronous operations with a retry limit
@@ -217,46 +215,14 @@ func CompileAndPublish(
 			pipelineFile = pipeline.GetData()
 		}
 
-		// send API call to capture repo for the counter (grabbing repo again to ensure counter is correct)
-		repo, err = database.GetRepoForOrg(ctx, r.GetOrg(), r.GetName())
-		if err != nil {
-			retErr := fmt.Errorf("%s: unable to get repo %s: %w", baseErr, r.GetFullName(), err)
-
-			// check if the retry limit has been exceeded
-			if i < cfg.Retries-1 {
-				logger.WithError(retErr).Warningf("retrying #%d", i+1)
-
-				// continue to the next iteration of the loop
-				continue
-			}
-
-			return nil, nil, http.StatusInternalServerError, retErr
-		}
-
-		// update DB record of repo (repo) with any changes captured from webhook payload (r)
-		repo.SetTopics(r.GetTopics())
-		repo.SetBranch(r.GetBranch())
-
-		// update the build numbers based off repo counter
-		inc := repo.GetCounter() + 1
-		repo.SetCounter(inc)
-		b.SetNumber(inc)
-
-		// populate the build link if a web address is provided
-		if len(cfg.Metadata.Vela.WebAddress) > 0 {
-			b.SetLink(
-				fmt.Sprintf("%s/%s/%d", cfg.Metadata.Vela.WebAddress, repo.GetFullName(), b.GetNumber()),
-			)
-		}
-
 		// ensure we use the expected pipeline type when compiling
 		//
 		// The pipeline type for a repo can change at any time which can break compiling
 		// existing pipelines in the system for that repo. To account for this, we update
 		// the repo pipeline type to match what was defined for the existing pipeline
 		// before compiling. After we're done compiling, we reset the pipeline type.
 		if len(pipeline.GetType()) > 0 {
-			repo.SetPipelineType(pipeline.GetType())
+			r.SetPipelineType(pipeline.GetType())
 		}
 
 		var compiled *types.Pipeline
@@ -268,7 +234,7 @@ func CompileAndPublish(
 			WithCommit(b.GetCommit()).
 			WithFiles(files).
 			WithMetadata(cfg.Metadata).
-			WithRepo(repo).
+			WithRepo(r).
 			WithUser(u).
 			WithLabels(cfg.Labels).
 			WithSCM(scm).
@@ -277,7 +243,7 @@ func CompileAndPublish(
 			Compile(ctx, pipelineFile)
 		if err != nil {
 			// format the error message with extra information
-			err = fmt.Errorf("unable to compile pipeline configuration for %s: %w", repo.GetFullName(), err)
+			err = fmt.Errorf("unable to compile pipeline configuration for %s: %w", r.GetFullName(), err)
 
 			// log the error for traceability
 			logger.Error(err.Error())
@@ -291,7 +257,7 @@ func CompileAndPublish(
 		// existing pipelines in the system for that repo. To account for this, we update
 		// the repo pipeline type to match what was defined for the existing pipeline
 		// before compiling. After we're done compiling, we reset the pipeline type.
-		repo.SetPipelineType(pipelineType)
+		r.SetPipelineType(pipelineType)
 
 		// skip the build if pipeline compiled to only the init and clone steps
 		skip := SkipEmptyBuild(p)
@@ -300,9 +266,9 @@ func CompileAndPublish(
 			b.SetStatus(constants.StatusSkipped)
 
 			// send API call to set the status on the commit using installation OR owner token
-			err = scm.Status(ctx, b, repo.GetOrg(), repo.GetName(), p.Token)
+			err = scm.Status(ctx, b, r.GetOrg(), r.GetName(), p.Token)
 			if err != nil {
-				logger.Errorf("unable to set commit status for %s/%d: %v", repo.GetFullName(), b.GetNumber(), err)
+				logger.Errorf("unable to set commit status for %s/%d: %v", r.GetFullName(), b.GetNumber(), err)
 			}
 
 			return nil,
@@ -316,7 +282,7 @@ func CompileAndPublish(
 		// validate deployment config
 		if (b.GetEvent() == constants.EventDeploy) && cfg.Deployment != nil {
 			if err := p.Deployment.Validate(cfg.Deployment.GetTarget(), cfg.Deployment.GetPayload()); err != nil {
-				retErr := fmt.Errorf("%s: failed to validate deployment for %s: %w", baseErr, repo.GetFullName(), err)
+				retErr := fmt.Errorf("%s: failed to validate deployment for %s: %w", baseErr, r.GetFullName(), err)
 
 				return nil, nil, http.StatusBadRequest, retErr
 			}
@@ -325,14 +291,14 @@ func CompileAndPublish(
 		// check if the pipeline did not already exist in the database
 		if pipeline == nil {
 			pipeline = compiled
-			pipeline.SetRepo(repo)
+			pipeline.SetRepo(r)
 			pipeline.SetCommit(b.GetCommit())
 			pipeline.SetRef(b.GetRef())
 
 			// send API call to create the pipeline
 			pipeline, err = database.CreatePipeline(ctx, pipeline)
 			if err != nil {
-				retErr := fmt.Errorf("%s: failed to create pipeline for %s: %w", baseErr, repo.GetFullName(), err)
+				retErr := fmt.Errorf("%s: failed to create pipeline for %s: %w", baseErr, r.GetFullName(), err)
 
 				// check if the retry limit has been exceeded
 				if i < cfg.Retries-1 {
@@ -347,9 +313,9 @@ func CompileAndPublish(
 
 			logger.WithFields(logrus.Fields{
 				"pipeline": pipeline.GetID(),
-				"org":      repo.GetOrg(),
-				"repo":     repo.GetName(),
-				"repo_id":  repo.GetID(),
+				"org":      r.GetOrg(),
+				"repo":     r.GetName(),
+				"repo_id":  r.GetID(),
 			}).Info("pipeline created")
 		}
 
@@ -362,7 +328,7 @@ func CompileAndPublish(
 		//   using the same Number and thus create a constraint
 		//   conflict; consider deleting the partially created
 		//   build object in the database
-		err = PlanBuild(ctx, database, scm, p, b, repo)
+		b, err = PlanBuild(ctx, database, scm, p, b, r)
 		if err != nil {
 			retErr := fmt.Errorf("%s: %w", baseErr, err)
 
@@ -382,42 +348,33 @@ func CompileAndPublish(
 			return nil, nil, http.StatusInternalServerError, retErr
 		}
 
+		// populate the build link if a web address is provided
+		if len(cfg.Metadata.Vela.WebAddress) > 0 {
+			b.SetLink(
+				fmt.Sprintf("%s/%s/%d", cfg.Metadata.Vela.WebAddress, r.GetFullName(), b.GetNumber()),
+			)
+		}
+
 		// break the loop because everything was successful
 		break
 	} // end of retry loop
 
-	// send API call to update repo for ensuring counter is incremented
-	repo, err = database.UpdateRepo(ctx, repo)
-	if err != nil {
-		retErr := fmt.Errorf("%s: failed to update repo %s: %w", baseErr, repo.GetFullName(), err)
-
-		return nil, nil, http.StatusInternalServerError, retErr
-	}
-
 	logger.WithFields(logrus.Fields{
-		"org":     repo.GetOrg(),
-		"repo":    repo.GetName(),
-		"repo_id": repo.GetID(),
+		"org":     r.GetOrg(),
+		"repo":    r.GetName(),
+		"repo_id": r.GetID(),
 	}).Info("repo updated - counter incremented")
 
 	// return error if pipeline didn't get populated
 	if p == nil {
-		retErr := fmt.Errorf("%s: failed to set pipeline for %s: %w", baseErr, repo.GetFullName(), err)
+		retErr := fmt.Errorf("%s: failed to set pipeline for %s: %w", baseErr, r.GetFullName(), err)
 
 		return nil, nil, http.StatusInternalServerError, retErr
 	}
 
 	// return error if build didn't get populated
 	if b == nil {
-		retErr := fmt.Errorf("%s: failed to set build for %s: %w", baseErr, repo.GetFullName(), err)
-
-		return nil, nil, http.StatusInternalServerError, retErr
-	}
-
-	// send API call to capture the triggered build
-	b, err = database.GetBuildForRepo(ctx, repo, b.GetNumber())
-	if err != nil {
-		retErr := fmt.Errorf("%s: failed to get new build %s/%d: %w", baseErr, repo.GetFullName(), b.GetNumber(), err)
+		retErr := fmt.Errorf("%s: failed to set build for %s: %w", baseErr, r.GetFullName(), err)
 
 		return nil, nil, http.StatusInternalServerError, retErr
 	}
@@ -438,7 +395,7 @@ func CompileAndPublish(
 	// publish the pipeline.Build to the build_executables table to be requested by a worker
 	err = PublishBuildExecutable(ctx, database, p, b)
 	if err != nil {
-		retErr := fmt.Errorf("unable to publish build executable for %s/%d: %w", repo.GetFullName(), b.GetNumber(), err)
+		retErr := fmt.Errorf("unable to publish build executable for %s/%d: %w", r.GetFullName(), b.GetNumber(), err)
 
 		return nil, nil, http.StatusInternalServerError, retErr
 	}

api/build/plan.go

@@ -22,12 +22,11 @@ import (
 // and services, for the build.
 // TODO:
 // - return build and error.
-func PlanBuild(ctx context.Context, database database.Interface, scm scm.Service, p *pipeline.Build, b *types.Build, r *types.Repo) error {
+func PlanBuild(ctx context.Context, database database.Interface, scm scm.Service, p *pipeline.Build, b *types.Build, r *types.Repo) (*types.Build, error) {
 	// update fields in build object
 	b.SetCreated(time.Now().UTC().Unix())
 
 	// send API call to create the build
-	// TODO: return created build and error instead of just error
 	b, err := database.CreateBuild(ctx, b)
 	if err != nil {
 		// clean up the objects from the pipeline in the database
@@ -39,7 +38,7 @@ func PlanBuild(ctx context.Context, database database.Interface, scm scm.Service
 		// - do we want to update the build or just delete it?
 		CleanBuild(ctx, database, b, nil, nil, err)
 
-		return fmt.Errorf("unable to create new build for %s: %w", r.GetFullName(), err)
+		return nil, fmt.Errorf("unable to create new build for %s: %w", r.GetFullName(), err)
 	}
 
 	logrus.WithFields(logrus.Fields{
@@ -53,7 +52,7 @@ func PlanBuild(ctx context.Context, database database.Interface, scm scm.Service
 		// clean up the objects from the pipeline in the database
 		CleanBuild(ctx, database, b, services, nil, err)
 
-		return err
+		return nil, err
 	}
 
 	// plan all steps for the build
@@ -62,8 +61,8 @@ func PlanBuild(ctx context.Context, database database.Interface, scm scm.Service
 		// clean up the objects from the pipeline in the database
 		CleanBuild(ctx, database, b, services, steps, err)
 
-		return err
+		return nil, err
 	}
 
-	return nil
+	return b, nil
 }

api/hook/create.go

@@ -85,30 +85,13 @@ func CreateHook(c *gin.Context) {
 		return
 	}
 
-	// send API call to capture the last hook for the repo
-	lastHook, err := database.FromContext(c).LastHookForRepo(ctx, r)
-	if err != nil {
-		retErr := fmt.Errorf("unable to get last hook for repo %s: %w", r.GetFullName(), err)
-
-		util.HandleError(c, http.StatusInternalServerError, retErr)
-
-		return
-	}
-
 	// update fields in webhook object
 	input.SetRepo(r)
-	input.SetNumber(1)
 
 	if input.GetCreated() == 0 {
 		input.SetCreated(time.Now().UTC().Unix())
 	}
 
-	if lastHook != nil {
-		input.SetNumber(
-			lastHook.GetNumber() + 1,
-		)
-	}
-
 	// send API call to create the webhook
 	h, err := database.FromContext(c).CreateHook(ctx, input)
 	if err != nil {

api/repo/create.go

@@ -261,8 +261,6 @@ func CreateRepo(c *gin.Context) {
 
 	// err being nil means we have a record of this repo (dbRepo)
 	if err == nil {
-		h, _ = database.FromContext(c).LastHookForRepo(ctx, dbRepo)
-
 		// make sure our record of the repo allowed events matches what we send to SCM
 		// what the dbRepo has should override default events on enable
 		r.SetAllowEvents(dbRepo.GetAllowEvents())
@@ -271,7 +269,7 @@ func CreateRepo(c *gin.Context) {
 	// check if we should create the webhook
 	if c.Value("webhookvalidation").(bool) {
 		// send API call to create the webhook
-		h, _, err = scm.FromContext(c).Enable(ctx, u, r, h)
+		h, _, err = scm.FromContext(c).Enable(ctx, u, r)
 		if err != nil {
 			retErr := fmt.Errorf("unable to create webhook for %s: %w", r.GetFullName(), err)
 

api/repo/repair.go

@@ -9,6 +9,7 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/sirupsen/logrus"
 
+	"github.com/go-vela/server/api/types"
 	wh "github.com/go-vela/server/api/webhook"
 	"github.com/go-vela/server/database"
 	"github.com/go-vela/server/internal"
@@ -63,7 +64,7 @@ import (
 // RepairRepo represents the API handler to remove
 // and then create a webhook for a repo.
 //
-//nolint:funlen // ignore statement count
+
 func RepairRepo(c *gin.Context) {
 	// capture middleware values
 	m := c.MustGet("metadata").(*internal.Metadata)
@@ -74,6 +75,8 @@ func RepairRepo(c *gin.Context) {
 
 	l.Debugf("repairing repo %s", r.GetFullName())
 
+	var hook *types.Hook
+
 	// check if we should create the webhook
 	if c.Value("webhookvalidation").(bool) {
 		// send API call to remove the webhook
@@ -86,17 +89,8 @@ func RepairRepo(c *gin.Context) {
 			return
 		}
 
-		hook, err := database.FromContext(c).LastHookForRepo(ctx, r)
-		if err != nil {
-			retErr := fmt.Errorf("unable to get last hook for %s: %w", r.GetFullName(), err)
-
-			util.HandleError(c, http.StatusInternalServerError, retErr)
-
-			return
-		}
-
 		// send API call to create the webhook
-		hook, _, err = scm.FromContext(c).Enable(ctx, u, r, hook)
+		hook, _, err = scm.FromContext(c).Enable(ctx, u, r)
 		if err != nil {
 			retErr := fmt.Errorf("unable to create webhook for %s: %w", r.GetFullName(), err)
 
@@ -143,22 +137,13 @@ func RepairRepo(c *gin.Context) {
 	// if repo has a name change, then update DB with new name
 	// if repo has an org change, update org as well
 	if sourceRepo.GetName() != r.GetName() || sourceRepo.GetOrg() != r.GetOrg() {
-		h, err := database.FromContext(c).LastHookForRepo(ctx, r)
-		if err != nil {
-			retErr := fmt.Errorf("unable to get last hook for %s: %w", r.GetFullName(), err)
-
-			util.HandleError(c, http.StatusInternalServerError, retErr)
-
-			return
-		}
-
 		// set sourceRepo PreviousName to old name if name is changed
 		// ignore if repo is transferred and name is unchanged
 		if sourceRepo.GetName() != r.GetName() {
 			sourceRepo.SetPreviousName(r.GetName())
 		}
 
-		r, err = wh.RenameRepository(ctx, l, database.FromContext(c), h, sourceRepo, r, m)
+		r, err = wh.RenameRepository(ctx, l, database.FromContext(c), sourceRepo, r, m)
 		if err != nil {
 			util.HandleError(c, http.StatusInternalServerError, err)
 			return

api/repo/update.go

@@ -225,7 +225,7 @@ func UpdateRepo(c *gin.Context) {
 	// if webhook validation is not set or events didn't change, skip webhook update
 	if c.Value("webhookvalidation").(bool) && eventsChanged {
 		// grab last hook from repo to fetch the webhook ID
-		lastHook, err := database.FromContext(c).LastHookForRepo(ctx, r)
+		lastHook, err := database.FromContext(c).GetHookForRepo(ctx, r, r.GetHookCounter())
 		if err != nil {
 			retErr := fmt.Errorf("unable to retrieve last hook for repo %s: %w", r.GetFullName(), err)
 

api/scm/sync.go

@@ -133,7 +133,7 @@ func SyncRepo(c *gin.Context) {
 	// update the repo hook in the SCM
 	if c.Value("webhookvalidation").(bool) && r.GetActive() {
 		// grab last hook from repo to fetch the webhook ID
-		lastHook, err := database.FromContext(c).LastHookForRepo(ctx, r)
+		lastHook, err := database.FromContext(c).GetHookForRepo(ctx, r, r.GetHookCounter())
 		if err != nil {
 			retErr := fmt.Errorf("unable to retrieve last hook for repo %s: %w", r.GetFullName(), err)
 

api/scm/sync_org.go

@@ -153,7 +153,7 @@ func SyncReposForOrg(c *gin.Context) {
 		// update the repo hook in the SCM
 		if c.Value("webhookvalidation").(bool) && repo.GetActive() {
 			// grab last hook from repo to fetch the webhook ID
-			lastHook, err := database.FromContext(c).LastHookForRepo(ctx, repo)
+			lastHook, err := database.FromContext(c).GetHookForRepo(ctx, repo, repo.GetHookCounter())
 			if err != nil {
 				retErr := fmt.Errorf("unable to retrieve last hook for repo %s: %w", repo.GetFullName(), err)